diff --git a/.github/workflows/reusable-code.yml b/.github/workflows/reusable-code.yml index d89289e110..2dd7d8e371 100644 --- a/.github/workflows/reusable-code.yml +++ b/.github/workflows/reusable-code.yml @@ -197,6 +197,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: code version: ${{ inputs.fullsend_version || job.workflow_sha }} diff --git a/.github/workflows/reusable-dispatch.yml b/.github/workflows/reusable-dispatch.yml index 1f2239322e..e6ef567e07 100644 --- a/.github/workflows/reusable-dispatch.yml +++ b/.github/workflows/reusable-dispatch.yml @@ -688,6 +688,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: triage fullsend-dir: ${{ inputs.install_mode == 'per-repo' && '.fullsend' || '' }} @@ -817,6 +818,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: code fullsend-dir: ${{ inputs.install_mode == 'per-repo' && '.fullsend' || '' }} @@ -936,6 +938,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: review fullsend-dir: ${{ inputs.install_mode == 'per-repo' && '.fullsend' || '' }} @@ -1206,6 +1209,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: fix fullsend-dir: ${{ inputs.install_mode == 'per-repo' && '.fullsend' || '' }} @@ -1308,6 +1312,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: retro fullsend-dir: ${{ inputs.install_mode == 'per-repo' && '.fullsend' || '' }} @@ -1395,6 +1400,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: prioritize fullsend-dir: ${{ inputs.install_mode == 'per-repo' && '.fullsend' || '' }} @@ -1720,6 +1726,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} JIRA_TOKEN: ${{ secrets.JIRA_TOKEN }} JIRA_USER_EMAIL: ${{ secrets.JIRA_USER_EMAIL }} JIRA_BASE_URL: ${{ inputs.jira_base_url || vars.JIRA_BASE_URL }} diff --git a/.github/workflows/reusable-fix.yml b/.github/workflows/reusable-fix.yml index 5d5737672e..e9d9df5e51 100644 --- a/.github/workflows/reusable-fix.yml +++ b/.github/workflows/reusable-fix.yml @@ -374,6 +374,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: fix version: ${{ inputs.fullsend_version || job.workflow_sha }} diff --git a/.github/workflows/reusable-prioritize.yml b/.github/workflows/reusable-prioritize.yml index f53e6d5803..1d152f8b6f 100644 --- a/.github/workflows/reusable-prioritize.yml +++ b/.github/workflows/reusable-prioritize.yml @@ -158,6 +158,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: prioritize version: ${{ inputs.fullsend_version || job.workflow_sha }} diff --git a/.github/workflows/reusable-retro.yml b/.github/workflows/reusable-retro.yml index e02a86bdb6..d86e283659 100644 --- a/.github/workflows/reusable-retro.yml +++ b/.github/workflows/reusable-retro.yml @@ -169,6 +169,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: retro version: ${{ inputs.fullsend_version || job.workflow_sha }} diff --git a/.github/workflows/reusable-review.yml b/.github/workflows/reusable-review.yml index 18983d99de..7cb12a9971 100644 --- a/.github/workflows/reusable-review.yml +++ b/.github/workflows/reusable-review.yml @@ -185,6 +185,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: review fullsend-dir: ${{ inputs.install_mode == 'per-repo' && '.fullsend' || '' }} diff --git a/.github/workflows/reusable-triage.yml b/.github/workflows/reusable-triage.yml index aa1a1f870b..d50879cba1 100644 --- a/.github/workflows/reusable-triage.yml +++ b/.github/workflows/reusable-triage.yml @@ -175,6 +175,7 @@ jobs: OTEL_EXPORTER_OTLP_CERTIFICATE: ${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }} OTEL_RESOURCE_ATTRIBUTES: ${{ vars.OTEL_RESOURCE_ATTRIBUTES }} OTEL_SDK_DISABLED: ${{ vars.OTEL_SDK_DISABLED }} + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: ${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }} with: agent: triage version: ${{ inputs.fullsend_version || job.workflow_sha }} diff --git a/docs/guides/dev/tracing.md b/docs/guides/dev/tracing.md index 4a9777eddc..3d5f85fc03 100644 --- a/docs/guides/dev/tracing.md +++ b/docs/guides/dev/tracing.md @@ -132,6 +132,43 @@ build the attribute slices. Start attributes: `iteration`, `exit_code`, `gen_ai.system`, model, token counts, `fullsend.cost_usd`, `fullsend.tool_calls`. +### Level 3 content on agent spans + +When the content-capture gate is on +(`telemetry.ContentCaptureEnabled()`), `runAgent` constructs one +`contentCollector` per iteration — iteration and agent span are 1:1, so a +run-scoped collector would repeat earlier iterations' content on later +spans — and tees the runtime's normalized event stream to it through +`RunParams.OnEvent`. + +**The tee trap:** supplying any `OnEvent` replaces the runtime's default +console renderer (`internal/runtime/claude.go`), so the handler built by +`contentEventHandler` always calls the renderer first and the collector +second. With the gate off the collector is nil and `contentEventHandler` +returns nil, leaving the default renderer path byte-identical to before +Level 3 existed. + +The collector (`internal/cli/content_collector.go`) coalesces contiguous +text/reasoning deltas, maps tool use to `tool_call` parts, redacts every +part through `security.OutputPipeline()` at assembly (redaction runs +before the size budget — truncating first could split a secret past +recognition), enforces a 256 KiB ordered-suffix budget (the ending survives — the +final answer is what consumers judge) with exact dropped-byte accounting +across content, tool names, and summaries, and emits +`gen_ai.output.messages` JSON following the GenAI output-messages schema, +including the schema-required `finish_reason` from the iteration outcome. `attachContent` records the +content and its marker attributes on the span before either +`finalizeAgentSpan` path can end it, so failed iterations keep their +content. + +**Consumer contract** (for eval scorers and other readers of +`run-telemetry.jsonl`): parse the `gen_ai.output.messages` attribute as +JSON; check `fullsend.content.truncated` / `fullsend.content.dropped_bytes` +before treating content as complete; masked secrets appear as the +redactor's mask tokens and are counted in `fullsend.content.redactions`. +The attribute names and shapes above are the consumption contract — see the +[Tracing reference](../infrastructure/distributed-tracing.md#content-capture-level-3). + ## Trace identity and TRACEPARENT propagation `resolveTraceIdentity()` handles W3C trace context propagation in three diff --git a/docs/guides/infrastructure/distributed-tracing.md b/docs/guides/infrastructure/distributed-tracing.md index 985905cfcf..1a6512ece6 100644 --- a/docs/guides/infrastructure/distributed-tracing.md +++ b/docs/guides/infrastructure/distributed-tracing.md @@ -12,10 +12,11 @@ For implementation details, see the |-------|-----------------|----------------------| | 1 | `run-telemetry.jsonl` file in the run output directory | None | | 2 | OTLP/HTTP export to a remote backend (metadata only) | `OTEL_EXPORTER_OTLP_*ENDPOINT` | -| 3 | Content capture (prompts, completions, tool I/O) in spans | `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true` *(planned, not yet implemented)* | +| 3 | Conversation content (assistant text, reasoning, tool calls) on `agent` spans | `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=true` | All levels produce metadata (timing, token counts, tool names, errors). -Level 3 adds prompt/completion content to spans. +Level 3 adds the agent's conversation content to spans — enabled by one +environment variable, exactly like Level 2's endpoint. ## Environment variables @@ -66,15 +67,52 @@ unset OTEL_EXPORTER_OTLP_TRACES_ENDPOINT | `TRACEPARENT` | W3C Trace Context parent | When present, the root span becomes `SpanKindConsumer`; when the sampled flag is unset (`-00`), OTLP export is suppressed but the local file is still written | | `TRACESTATE` | W3C Trace Context state | Propagated alongside `TRACEPARENT` | -### Content capture (planned) - -| Variable | Value | Effect | -|----------|-------|--------| -| `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` | `true` | Includes prompts, completions, tool arguments, tool results, and reasoning text in spans | - -Content capture follows the [OTel GenAI semantic conventions](https://github.com/open-telemetry/semantic-conventions/blob/v1.37.0/docs/gen-ai/gen-ai-spans.md). -When enabled, spans may contain proprietary source code, PII, or -credentials visible in tool outputs. +### Content capture (Level 3) + +Fullsend assembles Level 3 content from the normalized event stream the +console renders, redacts it through the security output pipeline, and +attaches it to the per-iteration `agent` span. The agent runtime's own +content-logging variables (`OTEL_LOG_USER_PROMPTS`, +`OTEL_LOG_ASSISTANT_RESPONSES`, etc.) are never set. + +| Variable | Values that enable capture | Values that keep it off | +|----------|---------------------------|-------------------------| +| `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` | `true`, `span_only`, `span_and_event` (case-insensitive) | unset, `false`, `NO_CONTENT`, `event_only`, anything unrecognized | + +The variable name and accepted values follow the +[OpenTelemetry GenAI instrumentation convention](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation-genai). +Fullsend records content on span attributes only, so `event_only` stays +off. An unrecognized value disables capture; telemetry never fails a run. + +**Captured:** assistant text, reasoning, and tool calls (name plus short +summary) — including any sub-agent activity, unattributed — as the +`gen_ai.output.messages` span attribute: a JSON string following the +[GenAI output-messages schema](https://github.com/open-telemetry/semantic-conventions/blob/v1.37.0/docs/gen-ai/gen-ai-output-messages.json) +with a `finish_reason` of `stop` or `error`. + +**Not captured:** model input (`gen_ai.input.messages`) and +pre/post-script content. First-iteration runs have no meaningful +runner-side input; retry iterations carry the injected validation +feedback, a natural input-capture follow-up. + +> **Planned:** Tool results, once a parser extension adds them to the +> normalized event stream — the next change after +> [#6429](https://github.com/fullsend-ai/fullsend/pull/6429). + +**Redaction and size:** every part passes through security redaction +(Unicode normalization, then secret masking) before reaching the span. +Content is bounded at 256 KiB per iteration, kept as an ordered suffix; +overflow drops the oldest content first. Truncation is marked via +`fullsend.content.truncated`. The SDK's span attribute length cap is +lifted while capture is on; an explicit +`OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` still wins and will cut content +mid-JSON — fullsend warns on stderr at startup. + +**Sinks:** content rides the span to both `run-telemetry.jsonl` and the +OTLP endpoint (when configured). Spans may contain proprietary source +code, PII, or credentials; the organization enabling capture is +responsible for its backend's access controls. For how MLflow displays +the content, see [Tracing with MLflow](../user/tracing-with-mlflow.md). ## Span hierarchy @@ -126,6 +164,10 @@ and are recognized by LLM-aware backends for GenAI dashboards. | `fullsend.prescript.skipped` | `run` | Whether the pre-script signaled a skip | | `fullsend.prescript.skip_reason` | `run` | Human-readable skip reason from the pre-script | | `fullsend.transcript_error` | `agent` | Present (`true`) when the agent exited 0 but its transcript reported an error — the span's status is Error while `exit_code` keeps the raw process exit | +| `gen_ai.output.messages` | `agent` | Level 3 only: the iteration's conversation content as a JSON string (see Content capture) | +| `fullsend.content.truncated` | `agent` | Level 3 only: present (`true`) when the size budget cut or dropped content | +| `fullsend.content.dropped_bytes` | `agent` | Level 3 only: exact content bytes removed by the size budget | +| `fullsend.content.redactions` | `agent` | Level 3 only: number of security findings raised while redacting content at assembly (including findings from parts the size budget later dropped) | ### Common attributes @@ -218,6 +260,7 @@ that hosts the fullsend caller workflows: | `OTEL_EXPORTER_OTLP_CERTIFICATE` | Variable | No | Path to a PEM CA bundle for backends behind a private CA. Commit the bundle into the config repo (e.g. `.fullsend/otel-ca.pem`) and set the variable to that checkout-relative path. | | `OTEL_RESOURCE_ATTRIBUTES` | Variable | No | Static `k=v,k=v` trace tags. The value is used verbatim; `${{ github.* }}` expressions evaluate only in workflow YAML, not in variables. | | `OTEL_SDK_DISABLED` | Variable | No | Set to `true` to disable all telemetry, including the local file exporter. | +| `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` | Variable | No | Set to `true` to attach conversation content to `agent` spans (Level 3; see Content capture). | Installations scaffolded before OTEL support was added must also forward the secrets (add `OTEL_EXPORTER_OTLP_TRACES_HEADERS` and @@ -237,6 +280,7 @@ env: OTEL_EXPORTER_OTLP_HEADERS: "${{ secrets.OTEL_EXPORTER_OTLP_HEADERS }}" OTEL_RESOURCE_ATTRIBUTES: "${{ vars.OTEL_RESOURCE_ATTRIBUTES }}" OTEL_SDK_DISABLED: "${{ vars.OTEL_SDK_DISABLED }}" + OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "${{ vars.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT }}" OTEL_EXPORTER_OTLP_CERTIFICATE: "${{ vars.OTEL_EXPORTER_OTLP_CERTIFICATE }}" ``` diff --git a/docs/guides/user/how-to-emit-traces.md b/docs/guides/user/how-to-emit-traces.md index 06fe3ac21b..0adf07f6ad 100644 --- a/docs/guides/user/how-to-emit-traces.md +++ b/docs/guides/user/how-to-emit-traces.md @@ -110,6 +110,18 @@ to a backend like MLflow, Jaeger, Grafana Tempo, etc. --org --repos repo1,repo2,repo3 ``` +## Capture conversation content + +Set `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` to add the agent's +text, reasoning, and tool calls to each `agent` span, in the local file and +at the endpoint. Content is redacted for secrets and bounded per iteration, +but may still contain proprietary code or PII — make sure your backend's +access controls fit before enabling it. + +```bash +gh variable set OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT --body "true" --repo +``` + ## Disable trace export Remove the endpoint variable and header secret from the repository or diff --git a/docs/guides/user/tracing-with-mlflow.md b/docs/guides/user/tracing-with-mlflow.md index f353c69425..1a49aac22a 100644 --- a/docs/guides/user/tracing-with-mlflow.md +++ b/docs/guides/user/tracing-with-mlflow.md @@ -80,6 +80,16 @@ excludes cache-creation and cache-read tokens, which dominate agent-run cost. The authoritative cost figure is the runtime-reported `fullsend.cost_usd` attribute on `agent` spans (also in `run-telemetry.jsonl`). +## Level 3 content + +With content capture enabled (see +[How To Emit Traces](how-to-emit-traces.md#capture-conversation-content)), +the conversation lives on each `agent` span as `gen_ai.output.messages`: +open the trace and select the span to read it. The trace list's +Request/Response preview columns derive from the root span only (capped +at 1000 characters), so they stay empty for fullsend traces — content is +deliberately not duplicated onto the root span. + ## Local development Start a local MLflow instance and point the exporter at it: diff --git a/docs/problems/security-threat-model.md b/docs/problems/security-threat-model.md index 32415194fa..cf77d4959b 100644 --- a/docs/problems/security-threat-model.md +++ b/docs/problems/security-threat-model.md @@ -467,3 +467,4 @@ Issue [#1685](https://github.com/fullsend-ai/fullsend/issues/1685) explores usin 6. **Immutable agent policy** — agent rules cannot be modified through the channels agents operate on 7. **No agent self-modification** — agents cannot change their own configuration, permissions, or system prompts 8. **Verify, don't trust** — system state must be checked independently of agent self-reports (see [agent self-report unreliability](#cross-cutting-concern-agent-self-report-unreliability)) +9. **Telemetry content boundary** — fullsend's content-handling guarantees (redaction at assembly, size bounds, opt-in gating) apply to its own extraction and redaction pipeline only; the agent runtime's native OTel instrumentation inside the sandbox is out of scope, like any other in-sandbox capability diff --git a/internal/cli/content_collector.go b/internal/cli/content_collector.go new file mode 100644 index 0000000000..b6adda6a1e --- /dev/null +++ b/internal/cli/content_collector.go @@ -0,0 +1,319 @@ +package cli + +import ( + "encoding/json" + "unicode/utf8" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" + + agentruntime "github.com/fullsend-ai/fullsend/internal/runtime" + "github.com/fullsend-ai/fullsend/internal/security" + "github.com/fullsend-ai/fullsend/internal/telemetry" +) + +// maxContentBytes bounds the conversation content attached to one agent +// span (one iteration), measured on the raw part bytes before JSON +// encoding. The value is far above what text+reasoning+tool-call +// summaries produce in practice (the full-transcript mean of ~369K chars +// includes tool results, which are not captured yet) and was accepted +// whole by the pilot backend in live validation. Revisit when tool +// results join the stream. +const maxContentBytes = 256 * 1024 + +// newContentCollectorIfEnabled returns a live collector when the Level 3 +// gate is on and nil otherwise — nil is the off state and is inert at +// every call site, so the gate needs no second check. +func newContentCollectorIfEnabled() *contentCollector { + if telemetry.ContentCaptureEnabled() { + return newContentCollector(maxContentBytes) + } + return nil +} + +// contentEventHandler tees the normalized event stream to the console +// renderer and the collector. A nil collector returns a nil handler so +// the runtime keeps its default renderer path — supplying any OnEvent +// replaces that renderer, and losing it silences CI output. +func contentEventHandler(render func(agentruntime.AgentEvent), c *contentCollector) func(agentruntime.AgentEvent) { + if c == nil { + return nil + } + return func(evt agentruntime.AgentEvent) { + render(evt) + c.Handle(evt) + } +} + +// attachContent records assembled content and its markers on the agent +// span. Markers attach even when the budget dropped every part — a +// consumer must always be able to tell partial from complete. The +// content value goes through stringAttr like every other dynamic +// attribute value (invalid UTF-8 in any string fails proto-marshal of +// the whole OTLP batch). +func attachContent(span trace.Span, res contentResult) { + attrs := make([]attribute.KeyValue, 0, 4) + if res.OutputMessages != "" { + attrs = append(attrs, stringAttr("gen_ai.output.messages", res.OutputMessages)) + } + if res.Truncated { + attrs = append(attrs, attribute.Bool("fullsend.content.truncated", true)) + } + if res.DroppedBytes > 0 { + attrs = append(attrs, attribute.Int("fullsend.content.dropped_bytes", res.DroppedBytes)) + } + if n := len(res.Findings); n > 0 { + attrs = append(attrs, attribute.Int("fullsend.content.redactions", n)) + } + if len(attrs) > 0 { + span.SetAttributes(attrs...) + } +} + +// contentPart is one part of the assembled assistant output message, +// shaped for the GenAI output-messages JSON schema: TextPart +// ({type:"text",content}), the schema's GenericPart extension point +// ({type:"reasoning",content}), and ToolCallRequestPart +// ({type:"tool_call",name}+summary). A tool summary is not the tool's +// arguments, so no arguments field is ever fabricated from it. +type contentPart struct { + Type string `json:"type"` + Content string `json:"content,omitempty"` + Name string `json:"name,omitempty"` + Summary string `json:"summary,omitempty"` +} + +func partSize(p contentPart) int { + return len(p.Content) + len(p.Name) + len(p.Summary) +} + +// contentMessage is one message in the gen_ai.output.messages array. +// finish_reason is REQUIRED by the schema's OutputMessage definition. +type contentMessage struct { + Role string `json:"role"` + Parts []contentPart `json:"parts"` + FinishReason string `json:"finish_reason"` +} + +// contentResult is the collector's assembled, redacted, size-bounded +// product, ready to attach to the iteration's agent span. +type contentResult struct { + // OutputMessages is the gen_ai.output.messages JSON string, empty + // when the iteration produced no content. + OutputMessages string + // DroppedBytes counts raw content bytes removed by the size budget + // (content, tool name, and summary bytes alike). + DroppedBytes int + // Truncated reports whether the budget cut or dropped anything. + Truncated bool + // Findings are the security findings raised during redaction. + Findings []security.Finding +} + +// contentCollector accumulates conversation content from the normalized +// AgentEvent stream — the same stream the console renders. It is +// runtime-agnostic by construction: it consumes only normalized event +// types, never a runtime's raw schema. A nil collector is the off state +// and every method is nil-safe. +// +// Redaction happens at assembly, because run-telemetry.jsonl is exempt +// from the host output scan — content that reaches a span is never swept +// afterwards. +type contentCollector struct { + maxBytes int + pipeline *security.Pipeline + parts []contentPart + total int + // evicted counts bytes of old parts discarded during accumulation. + // Eviction keeps memory bounded on long sessions by approximating the + // Result budget on sizes as accumulated — pre-redaction — so it can + // drop content the post-redaction suffix budget would have kept. + // Discarded content is always redacted first: its findings land in + // findings below, and no cut ever runs on raw bytes. + evicted int + // findings raised while redacting content during eviction; merged + // into the contentResult at Result so eviction-time redactions are + // counted exactly like assembly-time ones. + findings []security.Finding +} + +func newContentCollector(maxBytes int) *contentCollector { + return &contentCollector{maxBytes: maxBytes, pipeline: security.OutputPipeline()} +} + +// Handle consumes one normalized event. Contiguous text and reasoning +// deltas of the same kind coalesce into a single part (the Claude parser +// emits per-delta); tool calls are discrete parts. All other event kinds +// carry no conversation content and are ignored. +func (c *contentCollector) Handle(evt agentruntime.AgentEvent) { + if c == nil { + return + } + switch e := evt.(type) { + case agentruntime.TextEvent: + c.appendText("text", e.Text) + case agentruntime.ThinkingEvent: + c.appendText("reasoning", e.Text) + case agentruntime.ToolUseEvent: + c.parts = append(c.parts, contentPart{Type: "tool_call", Name: e.Name, Summary: e.Summary}) + c.total += len(e.Name) + len(e.Summary) + c.evictOverflow() + } +} + +func (c *contentCollector) appendText(kind, text string) { + if text == "" { + return + } + if n := len(c.parts); n > 0 && c.parts[n-1].Type == kind { + c.parts[n-1].Content += text + } else { + c.parts = append(c.parts, contentPart{Type: kind, Content: text}) + } + c.total += len(text) + c.evictOverflow() +} + +// evictOverflow discards accumulated content that the suffix budget +// would drop anyway, keeping memory bounded. Whole old parts go first; a +// single over-double-budget part has its head pre-trimmed. The +// redaction-before-truncation invariant holds here exactly as at Result: +// evicted parts are scanned before discard so their findings still +// count, and a pre-trim redacts first — cutting raw bytes could split a +// secret at the boundary so the redactor no longer recognizes the +// surviving fragment. Eviction decisions use pre-redaction sizes, so +// eviction approximates the Result budget and can drop content the +// post-redaction budget would have kept. Every evicted byte is counted +// so Result's accounting stays exact. +func (c *contentCollector) evictOverflow() { + for len(c.parts) > 1 { + head := partSize(c.parts[0]) + if c.total-head < c.maxBytes { + break + } + c.redact(c.parts[0].Content, &c.findings) + c.redact(c.parts[0].Name, &c.findings) + c.redact(c.parts[0].Summary, &c.findings) + c.evicted += head + c.total -= head + c.parts = c.parts[1:] + } + if len(c.parts) == 1 && c.parts[0].Type != "tool_call" && c.total > 2*c.maxBytes { + p := &c.parts[0] + before := len(p.Content) + p.Content = c.redact(p.Content, &c.findings) + c.total -= before - len(p.Content) + kept := tailToRuneBoundary(p.Content, c.maxBytes) + c.evicted += len(p.Content) - len(kept) + c.total -= len(p.Content) - len(kept) + p.Content = kept + } +} + +// Result assembles the redacted, size-bounded output messages for one +// iteration. finishReason is the schema-required outcome of the +// generation: "stop" for a normal finish, "error" when the iteration +// failed. Redaction runs before the size budget: truncating first could +// split a secret so the redactor no longer recognizes it. +func (c *contentCollector) Result(finishReason string) contentResult { + if c == nil || len(c.parts) == 0 { + return contentResult{} + } + + res := contentResult{ + DroppedBytes: c.evicted, + Truncated: c.evicted > 0, + // Findings raised while redacting evicted content count exactly + // like assembly-time ones — a consumer must see every redaction, + // including ones inside content the budget dropped. + Findings: append([]security.Finding(nil), c.findings...), + } + + redacted := make([]contentPart, 0, len(c.parts)) + for _, p := range c.parts { + p.Content = c.redact(p.Content, &res.Findings) + p.Name = c.redact(p.Name, &res.Findings) + p.Summary = c.redact(p.Summary, &res.Findings) + if partSize(p) == 0 { + continue // sanitized away entirely; the finding is recorded + } + redacted = append(redacted, p) + } + + // The budget keeps an ordered SUFFIX: the iteration's ending — the + // final answer — is what consumers judge, so overflow drops the + // oldest content first. The boundary part is tail-cut on a rune + // boundary (text/reasoning) or dropped whole (tool_call — a partial + // call would misrepresent it); everything older drops. + remaining := c.maxBytes + kept := make([]contentPart, 0, len(redacted)) + full := false + for i := len(redacted) - 1; i >= 0; i-- { + p := redacted[i] + size := partSize(p) + if !full && size <= remaining { + remaining -= size + kept = append(kept, p) + continue + } + res.Truncated = true + if !full && p.Type != "tool_call" && remaining > 0 { + tail := tailToRuneBoundary(p.Content, remaining) + res.DroppedBytes += size - len(tail) + if tail != "" { + p.Content = tail + kept = append(kept, p) + } + } else { + res.DroppedBytes += size + } + full = true + } + + if len(kept) == 0 { + return res + } + for i, j := 0, len(kept)-1; i < j; i, j = i+1, j-1 { + kept[i], kept[j] = kept[j], kept[i] + } + raw, err := json.Marshal([]contentMessage{{Role: "assistant", Parts: kept, FinishReason: finishReason}}) + if err != nil { + // Strings marshal unconditionally; treat the impossible as no content. + return res + } + res.OutputMessages = string(raw) + return res +} + +// redact runs text through the output pipeline, returning the sanitized +// form and accumulating findings. ScanResult.Sanitized is empty when +// nothing changed — but also when sanitization removed everything (an +// all-invisible-bytes input), so an empty Sanitized WITH findings means +// fully redacted, not unchanged. +func (c *contentCollector) redact(text string, findings *[]security.Finding) string { + if text == "" { + return text + } + scanned := c.pipeline.Scan(text) + *findings = append(*findings, scanned.Findings...) + if scanned.Sanitized != "" { + return scanned.Sanitized + } + if len(scanned.Findings) > 0 { + return "" + } + return text +} + +// tailToRuneBoundary keeps at most the last n bytes of s, starting on a +// rune boundary. +func tailToRuneBoundary(s string, n int) string { + if len(s) <= n { + return s + } + start := len(s) - n + for start < len(s) && !utf8.RuneStart(s[start]) { + start++ + } + return s[start:] +} diff --git a/internal/cli/content_collector_test.go b/internal/cli/content_collector_test.go new file mode 100644 index 0000000000..1904e63d64 --- /dev/null +++ b/internal/cli/content_collector_test.go @@ -0,0 +1,323 @@ +package cli + +import ( + "encoding/json" + "strings" + "testing" + "unicode/utf8" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + agentruntime "github.com/fullsend-ai/fullsend/internal/runtime" +) + +// decodeOutputMessages unmarshals a collector's OutputMessages JSON and +// asserts the structural requirements of the GenAI output-messages schema: +// an array of messages, each with a role, a parts array, and the +// REQUIRED finish_reason (the schema's OutputMessage.required is +// ["role","parts","finish_reason"]); every part carries a type. +func decodeOutputMessages(t *testing.T, raw string) []map[string]any { + t.Helper() + var msgs []map[string]any + require.NoError(t, json.Unmarshal([]byte(raw), &msgs), "OutputMessages must be valid JSON") + for _, m := range msgs { + require.Contains(t, m, "role", "schema requires role on every message") + require.Contains(t, m, "finish_reason", "schema requires finish_reason on every message") + parts, ok := m["parts"].([]any) + require.True(t, ok, "schema requires a parts array on every message") + for _, p := range parts { + part, ok := p.(map[string]any) + require.True(t, ok) + require.Contains(t, part, "type", "schema requires type on every part") + } + } + return msgs +} + +func partAt(t *testing.T, msgs []map[string]any, i int) map[string]any { + t.Helper() + require.NotEmpty(t, msgs) + parts := msgs[0]["parts"].([]any) + require.Greater(t, len(parts), i) + return parts[i].(map[string]any) +} + +func TestContentCollector_CoalescesContiguousDeltas(t *testing.T) { + c := newContentCollector(4096) + c.Handle(agentruntime.TextEvent{Text: "Hello "}) + c.Handle(agentruntime.TextEvent{Text: "world"}) + c.Handle(agentruntime.ThinkingEvent{Text: "pondering"}) + + res := c.Result("stop") + msgs := decodeOutputMessages(t, res.OutputMessages) + require.Len(t, msgs, 1) + assert.Equal(t, "assistant", msgs[0]["role"]) + assert.Equal(t, "stop", msgs[0]["finish_reason"]) + + text := partAt(t, msgs, 0) + assert.Equal(t, "text", text["type"]) + assert.Equal(t, "Hello world", text["content"]) + + reasoning := partAt(t, msgs, 1) + assert.Equal(t, "reasoning", reasoning["type"]) + assert.Equal(t, "pondering", reasoning["content"]) + + assert.Zero(t, res.DroppedBytes) + assert.False(t, res.Truncated) +} + +func TestContentCollector_FinishReasonError(t *testing.T) { + c := newContentCollector(4096) + c.Handle(agentruntime.TextEvent{Text: "partial answer before the crash"}) + + msgs := decodeOutputMessages(t, c.Result("error").OutputMessages) + assert.Equal(t, "error", msgs[0]["finish_reason"], + "a failed iteration's message must carry finish_reason=error") +} + +func TestContentCollector_ToolUseBecomesToolCallPart(t *testing.T) { + c := newContentCollector(4096) + c.Handle(agentruntime.ToolUseEvent{Name: "Bash", Summary: "ls -la"}) + + msgs := decodeOutputMessages(t, c.Result("stop").OutputMessages) + part := partAt(t, msgs, 0) + assert.Equal(t, "tool_call", part["type"]) + assert.Equal(t, "Bash", part["name"]) + assert.Equal(t, "ls -la", part["summary"]) + assert.NotContains(t, part, "arguments", + "a summary is not the tool's arguments; do not fabricate them") +} + +func TestContentCollector_RedactsSecretsAndSurfacesFindings(t *testing.T) { + secret := "ghp_" + strings.Repeat("a", 36) + c := newContentCollector(4096) + c.Handle(agentruntime.TextEvent{Text: "the token is " + secret}) + + res := c.Result("stop") + assert.NotContains(t, res.OutputMessages, secret, + "secret must not survive assembly") + assert.NotEmpty(t, res.Findings, + "a redaction hit must surface as a security finding") +} + +func TestContentCollector_RedactsToolCallNameAndSummary(t *testing.T) { + secret := "ghp_" + strings.Repeat("b", 36) + c := newContentCollector(4096) + c.Handle(agentruntime.ToolUseEvent{Name: "curl -H " + secret, Summary: "auth " + secret}) + + res := c.Result("stop") + assert.NotContains(t, res.OutputMessages, secret, + "tool_call name and summary are captured content and must be redacted") +} + +func TestContentCollector_PreservesCleanText(t *testing.T) { + c := newContentCollector(4096) + c.Handle(agentruntime.TextEvent{Text: "nothing secret here"}) + + res := c.Result("stop") + msgs := decodeOutputMessages(t, res.OutputMessages) + assert.Equal(t, "nothing secret here", partAt(t, msgs, 0)["content"], + "clean text must not be blanked by the empty-Sanitized convention") + assert.Empty(t, res.Findings) +} + +func TestContentCollector_SanitizedToEmptyDoesNotLeakRaw(t *testing.T) { + // All-null-byte text sanitizes to "", which collides with the + // pipeline's empty-Sanitized-means-unchanged convention. The raw + // bytes must NOT pass through; the finding must still be counted. + c := newContentCollector(4096) + c.Handle(agentruntime.TextEvent{Text: "\x00\x00\x00"}) + + res := c.Result("stop") + assert.NotContains(t, res.OutputMessages, "\\u0000", + "null bytes must not survive assembly") + assert.NotEmpty(t, res.Findings, "the normalizer's finding must be counted") +} + +func TestContentCollector_BudgetKeepsTheEnding(t *testing.T) { + // The budget keeps an ordered SUFFIX: the iteration's ending — the + // final answer — is what consumers judge, so overflow drops the + // oldest content first. + c := newContentCollector(40) + c.Handle(agentruntime.TextEvent{Text: strings.Repeat("early ", 40)}) + c.Handle(agentruntime.ThinkingEvent{Text: "last reasoning"}) + c.Handle(agentruntime.TextEvent{Text: "final answer"}) + + res := c.Result("stop") + assert.True(t, res.Truncated) + assert.Positive(t, res.DroppedBytes) + + msgs := decodeOutputMessages(t, res.OutputMessages) + parts := msgs[0]["parts"].([]any) + last := parts[len(parts)-1].(map[string]any) + assert.Equal(t, "final answer", last["content"], + "the ending must survive truncation intact") + for _, p := range parts { + content, _ := p.(map[string]any)["content"].(string) + assert.True(t, utf8.ValidString(content), "any cut must land on a rune boundary") + } +} + +func TestContentCollector_DroppedBytesAccountsExactly(t *testing.T) { + // Suffix policy on a single oversized part keeps the TAIL. A cut at + // 5 bytes from the end of "abcdéfgh" (9 bytes, é at [4:6]) lands at + // byte 4 — a rune start — keeping "éfgh" (5 bytes), dropping 4. + c := newContentCollector(5) + c.Handle(agentruntime.TextEvent{Text: "abcdéfgh"}) + + res := c.Result("stop") + require.True(t, res.Truncated) + msgs := decodeOutputMessages(t, res.OutputMessages) + kept := partAt(t, msgs, 0)["content"].(string) + assert.Equal(t, "éfgh", kept) + assert.Equal(t, 9-len(kept), res.DroppedBytes, + "DroppedBytes must be exactly original bytes minus kept bytes") +} + +func TestContentCollector_DroppedBytesCountToolCallNameBytes(t *testing.T) { + // A dropped tool_call part's budget footprint includes its Name — + // the docs define name+summary as captured content, so "exact + // dropped-byte accounting" must count both. + c := newContentCollector(10) + c.Handle(agentruntime.ToolUseEvent{Name: "0123456789ABCDEF", Summary: "0123456789"}) + c.Handle(agentruntime.TextEvent{Text: "final"}) + + res := c.Result("stop") + require.True(t, res.Truncated) + assert.Equal(t, 16+10, res.DroppedBytes, + "the dropped tool_call must account for name and summary bytes") + msgs := decodeOutputMessages(t, res.OutputMessages) + assert.Equal(t, "final", partAt(t, msgs, 0)["content"]) +} + +func TestContentCollector_NilIsInert(t *testing.T) { + var c *contentCollector + assert.NotPanics(t, func() { + c.Handle(agentruntime.TextEvent{Text: "x"}) + }) + assert.Empty(t, c.Result("stop").OutputMessages) +} + +func TestContentCollector_IgnoresNonContentEvents(t *testing.T) { + c := newContentCollector(4096) + c.Handle(agentruntime.InitEvent{Model: "m"}) + c.Handle(agentruntime.TokensEvent{InputTokens: 10}) + c.Handle(agentruntime.ResultEvent{NumTurns: 1}) + c.Handle(agentruntime.ErrorEvent{Message: "boom"}) + + res := c.Result("stop") + assert.Empty(t, res.OutputMessages, "no content events means no content") + assert.Zero(t, res.DroppedBytes) +} + +func TestContentCollector_EmptyDeltasProduceNoParts(t *testing.T) { + c := newContentCollector(4096) + c.Handle(agentruntime.TextEvent{Text: ""}) + + assert.Empty(t, c.Result("stop").OutputMessages) +} + +func TestContentCollector_MidRuneTailCutWalksForward(t *testing.T) { + // A cut at 4 bytes from the end of "abcdéfgh" lands inside é (bytes + // [4:6]); the boundary walk must move FORWARD, keeping "fgh" (3 + // bytes) and dropping 6 — never splitting the rune. + c := newContentCollector(4) + c.Handle(agentruntime.TextEvent{Text: "abcdéfgh"}) + + res := c.Result("stop") + msgs := decodeOutputMessages(t, res.OutputMessages) + kept := partAt(t, msgs, 0)["content"].(string) + assert.Equal(t, "fgh", kept) + assert.Equal(t, 9-len(kept), res.DroppedBytes) +} + +func TestContentCollector_PreTrimRedactsBeforeCutting(t *testing.T) { + // A secret straddling the eviction pre-trim boundary must be redacted + // BEFORE the head cut: trimming raw bytes first would split the secret + // so the redactor no longer recognizes the surviving fragment. With + // maxBytes=100 the deltas below total 250 (>2*100), and a raw-first + // trim would keep a 30-byte tail of the token verbatim. + // The tail after the secret uses "! " — characters outside the token + // alphabet — so the greedy token pattern cannot swallow it. + secret := "ghp_" + strings.Repeat("a", 36) + tail := strings.Repeat("! ", 35) + c := newContentCollector(100) + c.Handle(agentruntime.TextEvent{Text: strings.Repeat("x", 140)}) + c.Handle(agentruntime.TextEvent{Text: secret}) + c.Handle(agentruntime.TextEvent{Text: tail}) + + res := c.Result("stop") + assert.NotContains(t, res.OutputMessages, strings.Repeat("a", 10), + "no fragment of a boundary-straddling secret may survive the pre-trim") + assert.NotEmpty(t, res.Findings, + "the pre-trim redaction hit must surface as a security finding") + + msgs := decodeOutputMessages(t, res.OutputMessages) + kept := partAt(t, msgs, 0)["content"].(string) + assert.True(t, strings.HasSuffix(kept, tail), + "the ending must still survive the pre-trim") +} + +func TestContentCollector_EvictedPartsAreStillScanned(t *testing.T) { + // Whole parts evicted during accumulation never reach Result's redact + // pass, but their findings must still be counted — + // fullsend.content.redactions is documented to include findings from + // parts the size budget later dropped. + secret := "ghp_" + strings.Repeat("c", 36) + c := newContentCollector(30) + c.Handle(agentruntime.TextEvent{Text: "leak: " + secret}) + c.Handle(agentruntime.ThinkingEvent{Text: strings.Repeat("z", 30)}) + c.Handle(agentruntime.TextEvent{Text: "the end"}) + + require.LessOrEqual(t, len(c.parts), 2, + "the secret-bearing part must have been evicted during accumulation") + + res := c.Result("stop") + assert.NotContains(t, res.OutputMessages, secret) + assert.NotEmpty(t, res.Findings, + "findings inside evicted parts must still be counted") +} + +func TestContentCollector_EvictsWholeOldPartsExactly(t *testing.T) { + // Long sessions must not accumulate unbounded content: parts older + // than the suffix budget are evicted during Handle, and every + // evicted byte still lands in DroppedBytes. + c := newContentCollector(30) + c.Handle(agentruntime.ToolUseEvent{Name: "OldTool", Summary: strings.Repeat("x", 33)}) + c.Handle(agentruntime.ThinkingEvent{Text: strings.Repeat("y", 30)}) + c.Handle(agentruntime.TextEvent{Text: "the very end"}) + + require.LessOrEqual(t, len(c.parts), 2, "the old tool_call must have been evicted during accumulation") + + res := c.Result("stop") + require.True(t, res.Truncated) + msgs := decodeOutputMessages(t, res.OutputMessages) + parts := msgs[0]["parts"].([]any) + last := parts[len(parts)-1].(map[string]any) + assert.Equal(t, "the very end", last["content"]) + + kept := 0 + for _, p := range parts { + m := p.(map[string]any) + for _, k := range []string{"content", "name", "summary"} { + if v, ok := m[k].(string); ok { + kept += len(v) + } + } + } + assert.Equal(t, (7+33)+30+12-kept, res.DroppedBytes, + "evicted and budget-dropped bytes must sum exactly to original minus kept") +} + +func TestTailToRuneBoundary(t *testing.T) { + // The fits case is production-reachable: eviction pre-trims on + // post-redaction content, which masking can shrink under the bound. + assert.Equal(t, "fits", tailToRuneBoundary("fits", 10)) + assert.Equal(t, "fits", tailToRuneBoundary("fits", 4)) + assert.Equal(t, "", tailToRuneBoundary("anything", 0)) + assert.Equal(t, "défgh", tailToRuneBoundary("abcdéfgh", 6), + "cut landing on a rune start keeps the full tail") + assert.Equal(t, "fgh", tailToRuneBoundary("abcdéfgh", 4), + "cut landing mid-rune walks forward, never splitting the rune") +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 1f609588c5..53bcdf525e 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -1018,10 +1018,10 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep var runCount int tracer, tracingCleanup := telemetry.Setup(runDir, Version()) tid := resolveTraceIdentity(ctx, tracer, os.Getenv("TRACEPARENT"), os.Getenv("TRACESTATE"), []attribute.KeyValue{ - stringAttr("fullsend.agent", agentName), - stringAttr("fullsend.work_item_id", workItemID), + boundedStringAttr("fullsend.agent", agentName), + boundedStringAttr("fullsend.work_item_id", workItemID), attribute.String("gen_ai.operation.name", "invoke_agent"), - stringAttr("gen_ai.agent.name", agentName), + boundedStringAttr("gen_ai.agent.name", agentName), }) ctx = tid.Ctx rootSpan := tid.RootSpan @@ -1051,7 +1051,7 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep rootSpan.SetAttributes(attribute.Bool("fullsend.prescript.skipped", runSkipped)) } if runSkipped && runSkipReason != "" { - rootSpan.SetAttributes(stringAttr("fullsend.prescript.skip_reason", runSkipReason)) + rootSpan.SetAttributes(boundedStringAttr("fullsend.prescript.skip_reason", runSkipReason)) } if runCount > 0 { rootSpan.SetAttributes(rootSpanEndAttrs(aggMetrics, runCount)...) @@ -1622,6 +1622,10 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep go runHeartbeat(printer, agentStart, timeout, heartbeatDone) agentCtx, agentSpan := tracer.Start(ctx, "agent", trace.WithAttributes(agentSpanStartAttrs(iteration, agentName)...)) + // One collector per iteration: iteration and agent span are 1:1, so + // a run-scoped collector would repeat earlier iterations' content on + // later spans. Nil when the Level 3 gate is off; nil is inert. + collector := newContentCollectorIfEnabled() var metrics agentruntime.RunMetrics hooksSettings := "" if h.SecurityEnabled() { @@ -1641,13 +1645,29 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep Timeout: timeout, OutputPath: filepath.Join(iterDir, "output.jsonl"), Prompt: agentPrompt, + OnEvent: contentEventHandler(agentruntime.NewEventRenderer(printer).Handle, collector), }, printer, agentStart, &metrics) close(heartbeatDone) + // Attach content immediately before each finalize path ends the + // span, carrying the schema-required finish_reason from the + // iteration outcome. A failed iteration keeps its content — that + // is when the transcript matters most. attachContent no-ops on an + // empty result and attaches markers even when the budget dropped + // every part. + attachIterationContent := func(finishReason string) { + res := collector.Result(finishReason) + attachContent(agentSpan, res) + if n := len(res.Findings); n > 0 { + printer.StepWarn(fmt.Sprintf("Content capture redacted %d finding(s) from span content", n)) + } + } + // Accumulate behavioral metrics across iterations. aggregateRunMetrics(&aggMetrics, &metrics, iteration) if runErr != nil { + attachIterationContent("error") finalizeAgentSpan(agentSpan, runErr, iteration, exitCode, rt.System(), rt.Name(), &metrics, "") printer.StepFail("Agent execution failed") // Record the real exit code (rt.Run returns -1 when the agent never @@ -1685,6 +1705,16 @@ func runAgent(ctx context.Context, agentName, fullsendDir, outputBase, targetRep } } + // finish_reason reflects how the generation ended: a clean exit is + // "stop"; a non-zero exit or a transcript-reported error is "error" + // (both are schema enum values). Budget cuts are NOT "length" — + // that enum value means a model-side length stop, and telemetry + // cuts are marked by fullsend.content.truncated instead. + contentFinishReason := "stop" + if exitCode != 0 || transcriptErrMsg != "" { + contentFinishReason = "error" + } + attachIterationContent(contentFinishReason) finalizeAgentSpan(agentSpan, nil, iteration, exitCode, rt.System(), rt.Name(), &metrics, transcriptErrMsg) printer.Blank() @@ -2660,7 +2690,7 @@ func agentSpanStartAttrs(iteration int, agentName string) []attribute.KeyValue { return []attribute.KeyValue{ attribute.Int("iteration", iteration), attribute.String("gen_ai.operation.name", "invoke_agent"), - stringAttr("gen_ai.agent.name", agentName), + boundedStringAttr("gen_ai.agent.name", agentName), } } @@ -2678,7 +2708,7 @@ func agentSpanEndAttrs(iteration, exitCode int, system, runtimeName string, m *a attribute.Int("iteration", iteration), attribute.Int("exit_code", exitCode), stringAttr("gen_ai.system", system), - stringAttr("gen_ai.request.model", m.Model), + boundedStringAttr("gen_ai.request.model", m.Model), stringAttr("fullsend.runtime", runtimeName), attribute.Int("gen_ai.usage.input_tokens", m.InputTokens), attribute.Int("gen_ai.usage.output_tokens", m.OutputTokens), @@ -2839,6 +2869,16 @@ func stringAttr(key, val string) attribute.KeyValue { return attribute.String(key, strings.ToValidUTF8(val, "")) } +// boundedStringAttr is stringAttr plus a byte bound. Free-text attribute +// values that historically relied on the provider-wide SDK cap must use +// this: the Level 3 content gate lifts that cap (telemetry.spanLimits), so +// values from pre-script output, sandbox stream-json, or the environment +// would otherwise ride to the exporter unbounded and can get an oversized +// batch rejected whole. +func boundedStringAttr(key, val string) attribute.KeyValue { + return stringAttr(key, truncateStatusMsgTo(val, telemetry.MaxSpanAttrValueLen)) +} + // finalizeRootSpan records the run outcome on the root span and ends it: // a runtime error gets the bounded exception event before the status, and // the status comes from rootSpanStatus — validation, not the last agent @@ -3602,7 +3642,10 @@ func scanOutputFiles(outputDir, traceID string, printer *ui.Printer) error { } return nil } - // Skip the telemetry JSONL (metadata-only, still open for append). + // Skip the telemetry JSONL: it is still open for append, and any + // Level 3 conversation content in it was already redacted at + // assembly (contentCollector) before reaching a span, so it needs + // no post-hoc sweep. if path == filepath.Join(outputDir, telemetry.TelemetryFile) { return nil } diff --git a/internal/cli/scan_output_telemetry_test.go b/internal/cli/scan_output_telemetry_test.go index 13ee352028..eff57ddc90 100644 --- a/internal/cli/scan_output_telemetry_test.go +++ b/internal/cli/scan_output_telemetry_test.go @@ -16,8 +16,10 @@ import ( // TestScanOutputFiles_SkipsTelemetryArtifacts pins that the host-side output // redaction scan does NOT rewrite the telemetry JSONL file. It is still held // open for append during the scan, so an in-place rewrite would truncate it -// under the open handle; and it is metadata-only by construction. A normal -// output file must still be sanitized. +// under the open handle; and any Level 3 conversation content it carries was +// already redacted at assembly (contentCollector.Result) before reaching a +// span, so the file needs no post-hoc sweep. A normal output file must still +// be sanitized. func TestScanOutputFiles_SkipsTelemetryArtifacts(t *testing.T) { dir := t.TempDir() const secret = "Token: ghp_FAKEtesttoken000000000000000000000000\n" diff --git a/internal/cli/telemetry_run_test.go b/internal/cli/telemetry_run_test.go index 6240d25830..7cafb3b29c 100644 --- a/internal/cli/telemetry_run_test.go +++ b/internal/cli/telemetry_run_test.go @@ -2,8 +2,11 @@ package cli import ( "context" + "encoding/json" "errors" "fmt" + "os" + "path/filepath" "strings" "testing" "unicode/utf8" @@ -20,6 +23,7 @@ import ( "github.com/fullsend-ai/fullsend/internal/evalmeasure" agentruntime "github.com/fullsend-ai/fullsend/internal/runtime" "github.com/fullsend-ai/fullsend/internal/security" + "github.com/fullsend-ai/fullsend/internal/telemetry" ) func TestTelemetryExitCode(t *testing.T) { @@ -922,3 +926,209 @@ func TestTranscriptErrorMessage(t *testing.T) { assert.Equal(t, strings.Repeat(": ", 1999)+":"+"… (truncated)", got, "worst-case sanitized growth stays whole") } + +func TestAttachContent_SetsContentAndMarkerAttrs(t *testing.T) { + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + + _, span := tp.Tracer("test").Start(context.Background(), "agent") + attachContent(span, contentResult{ + OutputMessages: `[{"role":"assistant","parts":[{"type":"text","content":"hi"}]}]`, + DroppedBytes: 7, + Truncated: true, + Findings: []security.Finding{{Severity: "high"}}, + }) + span.End() + + ended := rec.Ended() + require.Len(t, ended, 1) + attrs := make(map[attribute.Key]attribute.Value) + for _, kv := range ended[0].Attributes() { + attrs[kv.Key] = kv.Value + } + require.Contains(t, attrs, attribute.Key("gen_ai.output.messages")) + assert.Contains(t, attrs[attribute.Key("gen_ai.output.messages")].AsString(), `"type":"text"`) + assert.Equal(t, int64(7), attrs[attribute.Key("fullsend.content.dropped_bytes")].AsInt64()) + assert.True(t, attrs[attribute.Key("fullsend.content.truncated")].AsBool()) + assert.Equal(t, int64(1), attrs[attribute.Key("fullsend.content.redactions")].AsInt64()) +} + +func TestAttachContent_EmptyResultAddsNothing(t *testing.T) { + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + + _, span := tp.Tracer("test").Start(context.Background(), "agent") + attachContent(span, contentResult{}) + span.End() + + ended := rec.Ended() + require.Len(t, ended, 1) + assert.Empty(t, ended[0].Attributes(), "no content must mean no attributes at all") +} + +func TestAttachContent_MarkersSurviveWhenAllContentDropped(t *testing.T) { + // A budget that drops every part must still leave its trace: the + // documented "a consumer can always tell partial from complete" + // invariant depends on the markers being attached even when no + // content attribute is. + rec := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + + _, span := tp.Tracer("test").Start(context.Background(), "agent") + attachContent(span, contentResult{Truncated: true, DroppedBytes: 9}) + span.End() + + ended := rec.Ended() + require.Len(t, ended, 1) + attrs := make(map[attribute.Key]attribute.Value) + for _, kv := range ended[0].Attributes() { + attrs[kv.Key] = kv.Value + } + assert.NotContains(t, attrs, attribute.Key("gen_ai.output.messages")) + assert.True(t, attrs[attribute.Key("fullsend.content.truncated")].AsBool()) + assert.Equal(t, int64(9), attrs[attribute.Key("fullsend.content.dropped_bytes")].AsInt64()) +} + +func TestBoundedStringAttr_CapsAtMaxSpanAttrValueLen(t *testing.T) { + // With the Level 3 gate lifting the provider-wide SDK cap, free-text + // attributes that used to rely on it must be bounded at the call + // site instead. + v := boundedStringAttr("k", strings.Repeat("x", telemetry.MaxSpanAttrValueLen*3)) + assert.LessOrEqual(t, len(v.Value.AsString()), telemetry.MaxSpanAttrValueLen) + + small := boundedStringAttr("k", "tiny") + assert.Equal(t, "tiny", small.Value.AsString()) +} + +func TestAgentSpanEndAttrs_ModelBoundedWithoutSDKCap(t *testing.T) { + // gen_ai.request.model comes from the sandboxed agent's stream-json + // output (untrusted, bounded only by the 1 MiB line buffer); it must + // not depend on the SDK cap the content gate lifts. + m := agentruntime.RunMetrics{Model: strings.Repeat("m", telemetry.MaxSpanAttrValueLen*2)} + for _, kv := range agentSpanEndAttrs(1, 0, "claude", "claude", &m) { + if kv.Key == "gen_ai.request.model" { + assert.LessOrEqual(t, len(kv.Value.AsString()), telemetry.MaxSpanAttrValueLen) + return + } + } + t.Fatal("gen_ai.request.model attribute not found") +} + +func TestContentEventHandler_NilCollectorKeepsDefaultRenderer(t *testing.T) { + assert.Nil(t, contentEventHandler(func(agentruntime.AgentEvent) {}, nil), + "gate off must leave OnEvent nil so the runtime's default renderer runs") +} + +func TestContentEventHandler_TeesToRendererAndCollector(t *testing.T) { + var rendered []agentruntime.AgentEvent + c := newContentCollector(4096) + handler := contentEventHandler(func(e agentruntime.AgentEvent) { rendered = append(rendered, e) }, c) + require.NotNil(t, handler) + + handler(agentruntime.TextEvent{Text: "hello"}) + handler(agentruntime.TokensEvent{InputTokens: 1}) + + assert.Len(t, rendered, 2, "every event must still reach the renderer") + assert.Contains(t, c.Result("stop").OutputMessages, "hello", "content events must reach the collector") +} + +func TestNewContentCollectorIfEnabled_FollowsGate(t *testing.T) { + t.Setenv(telemetry.ContentCaptureEnvVar, "") + assert.Nil(t, newContentCollectorIfEnabled(), "gate off => nil collector") + + t.Setenv(telemetry.ContentCaptureEnvVar, "true") + assert.NotNil(t, newContentCollectorIfEnabled(), "gate on => live collector") +} + +// TestContentCapture_EndToEndFileSink drives the REAL telemetry.Setup — +// file exporter plus the lifted attribute cap — and asserts that content +// larger than the metadata-mode 8192 cap survives byte-intact into +// run-telemetry.jsonl. This is the integration the unit tests cannot see: +// spanLimits and attachContent must agree or the SDK silently cuts the +// content JSON mid-value. +func TestContentCapture_EndToEndFileSink(t *testing.T) { + t.Setenv("OTEL_SDK_DISABLED", "") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") + t.Setenv("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") + t.Setenv(telemetry.ContentCaptureEnvVar, "true") + + dir := t.TempDir() + tracer, cleanup := telemetry.Setup(dir, "test") + + c := newContentCollectorIfEnabled() + require.NotNil(t, c) + big := strings.Repeat("all work and no play makes claude a dull agent. ", 400) // ~19KB > 8192 + c.Handle(agentruntime.TextEvent{Text: big}) + + _, span := tracer.Start(context.Background(), "agent") + res := c.Result("stop") + attachContent(span, res) + span.End() + cleanup(context.Background()) + + raw, err := os.ReadFile(filepath.Join(dir, telemetry.TelemetryFile)) + require.NoError(t, err) + require.NotEmpty(t, raw) + content := string(raw) + require.Contains(t, content, "gen_ai.output.messages") + assert.Contains(t, content, "dull agent", + "content must reach the file sink") + // The whole redacted content must survive — no SDK truncation at 8192. + var decoded []map[string]any + require.NoError(t, json.Unmarshal([]byte(res.OutputMessages), &decoded), + "assembled content must be valid JSON") + assert.Greater(t, len(res.OutputMessages), telemetry.MaxSpanAttrValueLen, + "fixture must exceed the metadata-mode cap for this test to prove anything") + escaped, err := json.Marshal(res.OutputMessages) + require.NoError(t, err) + assert.Contains(t, content, string(escaped[1:len(escaped)-1]), + "the full content JSON must appear byte-intact in the file sink") +} + +// TestContentCapture_GateOffProducesNoContent is the negative control: with +// the gate off the collector is nil, OnEvent stays nil (default renderer), +// and nothing content-shaped reaches the file sink. +func TestContentCapture_GateOffProducesNoContent(t *testing.T) { + t.Setenv("OTEL_SDK_DISABLED", "") + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "") + t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") + t.Setenv("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") + t.Setenv(telemetry.ContentCaptureEnvVar, "") + + dir := t.TempDir() + tracer, cleanup := telemetry.Setup(dir, "test") + + c := newContentCollectorIfEnabled() + require.Nil(t, c, "gate off must mean no collector") + c.Handle(agentruntime.TextEvent{Text: "would-be content"}) // nil-safe no-op + + _, span := tracer.Start(context.Background(), "agent") + attachContent(span, c.Result("stop")) + span.End() + cleanup(context.Background()) + + raw, err := os.ReadFile(filepath.Join(dir, telemetry.TelemetryFile)) + require.NoError(t, err) + assert.NotContains(t, string(raw), "gen_ai.output.messages") + assert.NotContains(t, string(raw), "would-be content") + assert.NotContains(t, string(raw), "fullsend.content.") +} + +func TestAgentSpanStartAttrs_AgentNameBoundedWithoutSDKCap(t *testing.T) { + // gen_ai.agent.name comes from the CLI argument (unbounded); like the + // other free-text attributes it must not depend on the SDK cap the + // content gate lifts. + for _, kv := range agentSpanStartAttrs(1, strings.Repeat("n", telemetry.MaxSpanAttrValueLen*2)) { + if kv.Key == "gen_ai.agent.name" { + assert.LessOrEqual(t, len(kv.Value.AsString()), telemetry.MaxSpanAttrValueLen) + return + } + } + t.Fatal("gen_ai.agent.name attribute not found") +} diff --git a/internal/scaffold/workflow_call_alignment_test.go b/internal/scaffold/workflow_call_alignment_test.go index 61f699b7b1..009124c969 100644 --- a/internal/scaffold/workflow_call_alignment_test.go +++ b/internal/scaffold/workflow_call_alignment_test.go @@ -431,6 +431,10 @@ func TestOTELVariableForwarding(t *testing.T) { "OTEL_EXPORTER_OTLP_CERTIFICATE", "OTEL_RESOURCE_ATTRIBUTES", "OTEL_SDK_DISABLED", + // Level 3 content-capture gate: a non-secret toggle, forwarded on + // the vars channel exactly like OTEL_SDK_DISABLED so orgs on managed + // workflows can enable it (ADR 0050 Level 3). + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", } forwardLine := func(v string) string { diff --git a/internal/telemetry/content.go b/internal/telemetry/content.go new file mode 100644 index 0000000000..50dce710dd --- /dev/null +++ b/internal/telemetry/content.go @@ -0,0 +1,35 @@ +package telemetry + +import ( + "os" + "strings" +) + +// ContentCaptureEnvVar is the Level 3 content-capture opt-in named by +// ADR 0050. The variable name and its value vocabulary come from the +// OpenTelemetry GenAI instrumentation convention (documented by the +// opentelemetry-python-contrib GenAI instrumentations); the pinned +// semantic-conventions v1.37.0 release does not define this variable. +// Fullsend's runner is the GenAI instrumentation that reads it — it is +// never passed to the agent runtime, whose own content-logging variables +// (OTEL_LOG_*) fullsend never sets. +const ContentCaptureEnvVar = "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT" + +// ContentCaptureEnabled reports whether Level 3 content capture is on. +// +// Fullsend records content on span attributes only, so the affirmative +// values are the ones whose intent span recording can honor: "true" +// (legacy boolean form), "span_only", and "span_and_event". "event_only" +// is off — fullsend implements no event-based capture, and recording on +// spans would contradict the operator's "only". "NO_CONTENT", "false", +// unset, and anything unrecognized are off: telemetry never fails a run +// (ADR 0050), so an unexpected value disables capture rather than +// erroring. +func ContentCaptureEnabled() bool { + switch strings.ToLower(strings.TrimSpace(os.Getenv(ContentCaptureEnvVar))) { + case "true", "span_only", "span_and_event": + return true + default: + return false + } +} diff --git a/internal/telemetry/content_test.go b/internal/telemetry/content_test.go new file mode 100644 index 0000000000..f7e80fc9a8 --- /dev/null +++ b/internal/telemetry/content_test.go @@ -0,0 +1,69 @@ +package telemetry + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestSpanLimits_ContentCaptureLiftsDefaultCap(t *testing.T) { + cases := []struct { + name string + gate string + limitEnv string + wantLimit int + }{ + // Gate off: the #5944 default cap stands. + {"gate off keeps default", "", "", MaxSpanAttrValueLen}, + // Gate on, no operator limit: unlimited — the SDK cap would cut the + // content JSON mid-value, producing invalid JSON; the collector's + // byte budget is the size bound instead. + {"gate on lifts cap", "true", "", -1}, + // An operator's explicit limit always wins, gate or no gate. + {"operator limit wins over gate", "true", "512", 512}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pinOTELEnv(t) + t.Setenv(ContentCaptureEnvVar, tc.gate) + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", tc.limitEnv) + assert.Equal(t, tc.wantLimit, spanLimits().AttributeValueLengthLimit) + }) + } +} + +func TestContentCaptureEnabled_ValueContract(t *testing.T) { + cases := []struct { + name string + value string + set bool + want bool + }{ + {"unset", "", false, false}, + {"set but empty", "", true, false}, + {"true", "true", true, true}, + {"TRUE uppercase", "TRUE", true, true}, + {"padded true", " true ", true, true}, + {"span_only", "span_only", true, true}, + {"SPAN_ONLY uppercase", "SPAN_ONLY", true, true}, + {"span_and_event", "span_and_event", true, true}, + {"SPAN_AND_EVENT uppercase", "SPAN_AND_EVENT", true, true}, + // fullsend records content on span attributes only. event_only asks + // for capture exclusively on events, which fullsend cannot honor — + // recording on spans would contradict the operator's "only". + {"event_only is off", "event_only", true, false}, + {"EVENT_ONLY uppercase is off", "EVENT_ONLY", true, false}, + {"NO_CONTENT is off", "NO_CONTENT", true, false}, + {"no_content lowercase is off", "no_content", true, false}, + {"false is off", "false", true, false}, + {"unrecognized is off", "yes-please", true, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if tc.set { + t.Setenv(ContentCaptureEnvVar, tc.value) + } + assert.Equal(t, tc.want, ContentCaptureEnabled()) + }) + } +} diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index a9290710a5..d8d5a300d9 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -79,8 +79,11 @@ func validateEndpoints(endpoint, tracesEndpoint string) error { return nil } -// MaxSpanAttrValueLen bounds every span attribute value recorded through -// this provider. The SDK applies the limit to span attributes only — +// MaxSpanAttrValueLen bounds span attribute values recorded through this +// provider in metadata-only mode. When the Level 3 content gate is on, +// spanLimits lifts the provider-wide cap (a capped cut would corrupt the +// content JSON mid-value), so free-text values that relied on this cap +// are bounded at their call sites instead (internal/cli boundedStringAttr). The SDK applies the limit to span attributes only — // event messages are bounded at their call site — counting characters, // not bytes (a multibyte value can reach four bytes per character on the // wire), and it repairs invalid UTF-8 only when it truncates: values at @@ -104,27 +107,65 @@ const MaxSpanAttrValueLen = 8192 func spanLimits() sdktrace.SpanLimits { limits := sdktrace.NewSpanLimits() if limits.AttributeValueLengthLimit < 0 && !attrValueLenConfigured() { + if ContentCaptureEnabled() { + // Level 3 puts JSON-string content attributes on spans. The + // default cap would cut such a value mid-string and corrupt + // the JSON; the content collector's byte budget is the size + // bound, so the SDK cap stays unlimited. An operator's + // explicit limit env var still wins above. + return limits + } limits.AttributeValueLengthLimit = MaxSpanAttrValueLen } return limits } // attrValueLenConfigured reports whether the SDK honored an operator's -// attribute value-length limit. It mirrors the SDK's firstInt resolution +// attribute value-length limit. +func attrValueLenConfigured() bool { + _, ok := operatorAttrValueLimit() + return ok +} + +// operatorAttrValueLimit resolves the operator's attribute value-length +// limit env vars to the value the SDK honored, reporting ok=false when no +// operator setting took effect. It mirrors the SDK's firstInt resolution // exactly (sdk/trace/internal/env, v1.44.0): the first non-empty variable // decides alone — if its value fails strconv.Atoi, the SDK falls back to // its default without consulting the second variable, so a discarded // override is not a setting here either. -func attrValueLenConfigured() bool { +func operatorAttrValueLimit() (int, bool) { for _, key := range []string{"OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT"} { v := os.Getenv(key) if v == "" { continue } - _, err := strconv.Atoi(v) - return err == nil + n, err := strconv.Atoi(v) + return n, err == nil + } + return 0, false +} + +// warnContentCaptureAttrLimit warns on stderr when the Level 3 content +// gate is on but an operator's finite attribute value-length limit is +// configured. The operator limit wins over the gate's cap lift +// (spanLimits), so the SDK will cut any gen_ai.output.messages value over +// the limit mid-JSON — in both sinks, with no fullsend.content.truncated +// marker (that marker reflects only collector-side cuts) — silently +// breaking the documented consumer contract. Telemetry never fails a run +// (ADR 0050), so the collision is surfaced, not fatal. An explicit -1 +// (unlimited) cannot cut and is not a conflict. +func warnContentCaptureAttrLimit() { + if !ContentCaptureEnabled() { + return + } + if limit, ok := operatorAttrValueLimit(); ok && limit >= 0 { + fmt.Fprintf(os.Stderr, + "fullsend: content capture is enabled but the operator attribute value length limit (%d) is set; "+ + "gen_ai.output.messages values over the limit will be cut mid-JSON (unparseable, and "+ + "fullsend.content.truncated will not flag the cut) — raise the limit or unset it to keep content parseable\n", + limit) } - return false } // Setup creates a TracerProvider with file and (optionally) OTLP exporters. @@ -145,6 +186,8 @@ func Setup(dir string, serviceVersion string) (trace.Tracer, func(context.Contex return tracenoop.NewTracerProvider().Tracer(""), noop } + warnContentCaptureAttrLimit() + res := buildResource(serviceVersion) opts := []sdktrace.TracerProviderOption{ sdktrace.WithResource(res), diff --git a/internal/telemetry/telemetry_test.go b/internal/telemetry/telemetry_test.go index aee8daf547..e69b338121 100644 --- a/internal/telemetry/telemetry_test.go +++ b/internal/telemetry/telemetry_test.go @@ -40,6 +40,7 @@ func pinOTELEnv(t *testing.T) { t.Setenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "") t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") t.Setenv("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", "") + t.Setenv(ContentCaptureEnvVar, "") } func TestSetup_FileExporter(t *testing.T) { @@ -192,6 +193,66 @@ func TestSpanLimits(t *testing.T) { "a valid specific var wins regardless of the generic one") } +// TestSetup_ContentCaptureOperatorLimitWarning pins the collision warning: +// when the Level 3 gate is on but an operator's finite attribute value +// length limit is configured, the SDK will cut gen_ai.output.messages +// mid-JSON (no fullsend.content.truncated marker reflects an SDK cut), so +// Setup must say so on stderr instead of letting the contract break +// silently. An explicit -1 (unlimited) is not a conflict. +func TestSetup_ContentCaptureOperatorLimitWarning(t *testing.T) { + cases := []struct { + name string + gate string + specific string // OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT + generic string // OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT + want bool + }{ + {"gate on with finite specific limit warns", "true", "512", "", true}, + {"gate on with finite generic limit warns", "true", "", "8192", true}, + {"gate on with explicit unlimited does not warn", "true", "-1", "", false}, + {"gate on with no limit does not warn", "true", "", "", false}, + {"gate off with finite limit does not warn", "", "512", "", false}, + {"gate on with unparseable limit does not warn", "true", "garbage", "", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pinOTELEnv(t) + t.Setenv(ContentCaptureEnvVar, tc.gate) + t.Setenv("OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT", tc.specific) + t.Setenv("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", tc.generic) + + pr, pw, err := os.Pipe() + require.NoError(t, err) + oldStderr := os.Stderr + os.Stderr = pw + defer func() { os.Stderr = oldStderr }() + + var captured []byte + done := make(chan struct{}) + go func() { + captured, _ = io.ReadAll(pr) + close(done) + }() + + dir := t.TempDir() + _, cleanup := Setup(dir, "1.0.0") + cleanup(context.Background()) + + os.Stderr = oldStderr + pw.Close() + <-done + + if tc.want { + assert.Contains(t, string(captured), "fullsend: content capture is enabled but the operator attribute value length limit", + "Setup must warn when the operator limit will cut content JSON mid-value") + } else { + assert.NotContains(t, string(captured), "content capture", + "Setup must not warn when the configuration cannot cut content JSON") + } + }) + } +} + func TestSetup_NoopOnBadDir(t *testing.T) { pinOTELEnv(t) tracer, cleanup := Setup("/nonexistent/path/that/should/fail", "1.0.0")