From bf943acefc45ec1c368cdf90a8469c236ae668b0 Mon Sep 17 00:00:00 2001 From: Chad Hietala Date: Fri, 7 Aug 2026 15:39:36 -0400 Subject: [PATCH] feat(eve): declare content capture per trace destination MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `otelIntegration()` takes `recordInputs` and `recordOutputs`. Content is written onto the span if any destination wants it, and a destination that declined drops it on its way out through a processor in front of its own — so a local spool and a hosted backend no longer have to agree, while an agent whose every destination declines still never materializes a prompt. The redactor copies rather than strips in place: the span it is handed is shared with every other processor in the pipeline. `EVE_TRACES_CONTENT` becomes `localTraces()`'s override and only ever narrows, rather than deciding what a hosted backend beside it receives. Signed-off-by: Chad Hietala --- .changeset/olive-otters-attack.md | 9 + .../eve/src/public/instrumentation/otel.ts | 19 +- .../src/tracing/agent-otel-provider.test.ts | 4 +- .../eve/src/tracing/agent-otel-provider.ts | 6 +- .../src/tracing/content-attributes.test.ts | 73 ++++++ .../eve/src/tracing/content-attributes.ts | 33 ++- .../tracing/content-span-processor.test.ts | 244 ++++++++++++++++-- .../eve/src/tracing/content-span-processor.ts | 118 ++++++++- .../tracing/local-instrumentation-runtime.ts | 22 +- packages/eve/src/tracing/local-traces.ts | 11 +- .../eve/src/tracing/otel-declaration.test.ts | 36 +++ packages/eve/src/tracing/otel-declaration.ts | 57 ++-- 12 files changed, 550 insertions(+), 82 deletions(-) create mode 100644 .changeset/olive-otters-attack.md create mode 100644 packages/eve/src/tracing/content-attributes.test.ts diff --git a/.changeset/olive-otters-attack.md b/.changeset/olive-otters-attack.md new file mode 100644 index 0000000000..dbb89719c2 --- /dev/null +++ b/.changeset/olive-otters-attack.md @@ -0,0 +1,9 @@ +--- +"eve": patch +--- + +Content capture is now declared per trace destination. `otelIntegration()` takes +`recordInputs` and `recordOutputs`, so a local spool and a hosted backend no +longer have to agree on whether they see prompts and tool results — a +destination that declines never exports them. `EVE_TRACES_CONTENT=off` now +narrows `localTraces()` alone rather than the whole process. diff --git a/packages/eve/src/public/instrumentation/otel.ts b/packages/eve/src/public/instrumentation/otel.ts index 864db03dbb..87e9608a12 100644 --- a/packages/eve/src/public/instrumentation/otel.ts +++ b/packages/eve/src/public/instrumentation/otel.ts @@ -10,7 +10,6 @@ */ import { createLocalTracesProcessor, resolveLocalTracesContent } from "#tracing/local-traces.js"; -import { contentFilteringProcessor } from "#tracing/content-span-processor.js"; import { agentRunsIntegration, otelIntegration, @@ -43,17 +42,13 @@ export function agentRuns(options: ContentOptions = {}): OtelIntegration { * Export it from `agent/instrumentation/local.ts` to keep it alongside a hosted * backend, or export `disableInstrumentation()` from that file to turn it off. * Omitting the file leaves eve's default in place. - * `EVE_TRACES_CONTENT=off` narrows this destination only. + * + * `EVE_TRACES_CONTENT=off` narrows this destination and no other, so declining + * content locally leaves what a hosted backend receives alone. */ export function localTraces(options: ContentOptions = {}): OtelIntegration { - const content = resolveLocalTracesContent(options); - const spool = createLocalTracesProcessor(); - return { - ...otelIntegration(), - content, - spanProcessors: - content.recordInputs && content.recordOutputs - ? [spool] - : [contentFilteringProcessor(spool, content)], - }; + return otelIntegration({ + ...resolveLocalTracesContent(options), + spanProcessors: [createLocalTracesProcessor()], + }); } diff --git a/packages/eve/src/tracing/agent-otel-provider.test.ts b/packages/eve/src/tracing/agent-otel-provider.test.ts index 9b24609c8e..04d2bdfd21 100644 --- a/packages/eve/src/tracing/agent-otel-provider.test.ts +++ b/packages/eve/src/tracing/agent-otel-provider.test.ts @@ -450,10 +450,10 @@ describe("createAgentOtelInstrumentation", () => { spanProcessors: [new SimpleSpanProcessor(exporter)], }); const agentOtel = createAgentOtelInstrumentation({ - recordInputs: false, - recordOutputs: false, frameworkVersion: "test", idGenerator, + recordInputs: false, + recordOutputs: false, stateStore: new InMemoryAgentTraceStateStore(), tracer: provider.getTracer("eve.agent"), }); diff --git a/packages/eve/src/tracing/agent-otel-provider.ts b/packages/eve/src/tracing/agent-otel-provider.ts index 2607cc44bb..f0e76412b0 100644 --- a/packages/eve/src/tracing/agent-otel-provider.ts +++ b/packages/eve/src/tracing/agent-otel-provider.ts @@ -58,10 +58,12 @@ interface ToolSpanState extends SpanState { export interface AgentOtelInstrumentationInput { /** - * The union of what this process's destinations requested. Each destination - * independently removes declined content before export. + * Whether to write model prompts and tool call inputs onto spans at all. + * This is the union across destinations, not one destination's policy: a + * destination that declined drops these on its way out instead. */ readonly recordInputs?: boolean; + /** The same, for model responses and tool call outputs. */ readonly recordOutputs?: boolean; readonly frameworkVersion: string; readonly idGenerator: AgentSpanIdGenerator; diff --git a/packages/eve/src/tracing/content-attributes.test.ts b/packages/eve/src/tracing/content-attributes.test.ts new file mode 100644 index 0000000000..edacb0268b --- /dev/null +++ b/packages/eve/src/tracing/content-attributes.test.ts @@ -0,0 +1,73 @@ +import { describe, expect, it } from "vitest"; + +import { withoutDeclinedContent } from "#tracing/content-attributes.js"; + +const ATTRIBUTES = { + "ai.prompt.messages": "what the user said", + "ai.response.finish_reason": "stop", + "ai.response.text": "what the model said", + "gen_ai.request.model": "test-model", + "gen_ai.tool.call.arguments": "{}", + "gen_ai.tool.call.result": "42", + "gen_ai.tool.name": "weather", +}; + +describe("withoutDeclinedContent", () => { + it("keeps everything when the destination declined nothing", () => { + expect( + withoutDeclinedContent(ATTRIBUTES, { recordInputs: true, recordOutputs: true }), + ).toBeUndefined(); + }); + + it("keeps everything when the span carries none of what was declined", () => { + expect( + withoutDeclinedContent( + { "gen_ai.request.model": "test-model" }, + { recordInputs: false, recordOutputs: false }, + ), + ).toBeUndefined(); + }); + + it("drops inputs alone", () => { + expect( + withoutDeclinedContent(ATTRIBUTES, { recordInputs: false, recordOutputs: true }), + ).toEqual({ + "ai.response.finish_reason": "stop", + "ai.response.text": "what the model said", + "gen_ai.request.model": "test-model", + "gen_ai.tool.call.result": "42", + "gen_ai.tool.name": "weather", + }); + }); + + it("drops outputs alone", () => { + expect( + withoutDeclinedContent(ATTRIBUTES, { recordInputs: true, recordOutputs: false }), + ).toEqual({ + "ai.prompt.messages": "what the user said", + "ai.response.finish_reason": "stop", + "gen_ai.request.model": "test-model", + "gen_ai.tool.call.arguments": "{}", + "gen_ai.tool.name": "weather", + }); + }); + + // The prefixes are shared: `ai.response.finish_reason` and `gen_ai.tool.name` + // say what happened rather than what was said, so declining content cannot + // cost a destination the ability to read its own traces. + it("keeps metadata that shares a prefix with content", () => { + expect( + withoutDeclinedContent(ATTRIBUTES, { recordInputs: false, recordOutputs: false }), + ).toEqual({ + "ai.response.finish_reason": "stop", + "gen_ai.request.model": "test-model", + "gen_ai.tool.name": "weather", + }); + }); + + it("leaves the attributes it was handed alone", () => { + const attributes = { ...ATTRIBUTES }; + withoutDeclinedContent(attributes, { recordInputs: false, recordOutputs: false }); + expect(attributes).toEqual(ATTRIBUTES); + }); +}); diff --git a/packages/eve/src/tracing/content-attributes.ts b/packages/eve/src/tracing/content-attributes.ts index 7f4657eb31..a31cdc4edd 100644 --- a/packages/eve/src/tracing/content-attributes.ts +++ b/packages/eve/src/tracing/content-attributes.ts @@ -1,4 +1,19 @@ -/** Prompts, instructions, and tool arguments. */ +/** + * Which span attributes carry conversation content, and in which direction. + * + * Two vocabularies land on the same spans: the ones eve sets on its own + * `agent.*` spans, and the ones the AI SDK's OpenTelemetry integration sets on + * the model-call spans beneath them. A destination that declined content has to + * be rid of both, so both are listed here rather than in the module that writes + * each. + * + * Listed by name rather than by prefix because the prefixes are shared with + * metadata that must survive: `ai.response.finish_reason` and + * `gen_ai.tool.name` say what happened, not what was said. The cost is that a + * new content attribute in a dependency is not covered until it is added here. + */ + +/** Prompts, instructions, tool arguments — what went in. */ const INPUT_CONTENT_ATTRIBUTES: ReadonlySet = new Set([ "ai.documents", "ai.prompt", @@ -14,7 +29,12 @@ const INPUT_CONTENT_ATTRIBUTES: ReadonlySet = new Set([ "gen_ai.tool.definitions", ]); -/** Responses, reasoning, and tool results. */ +/** + * Responses, reasoning, tool results — what came out. + * + * `ai.toolCall.args` is here rather than above because the AI SDK gates it on + * `recordOutputs`; matching that is what keeps one destination's view coherent. + */ const OUTPUT_CONTENT_ATTRIBUTES: ReadonlySet = new Set([ "ai.embedding", "ai.embeddings", @@ -43,7 +63,14 @@ function isDeclined(key: string, content: ResolvedContentOptions): boolean { return !content.recordOutputs && OUTPUT_CONTENT_ATTRIBUTES.has(key); } -/** Returns a copy without declined content, or undefined when no copy is needed. */ +/** + * The attributes with the declined content removed, or `undefined` when there + * was none to remove. + * + * The `undefined` return is what lets the caller forward the original span + * untouched in the common case, so a destination that declined a direction the + * span never carried costs one pass over its keys and no allocation. + */ export function withoutDeclinedContent( attributes: Readonly>, content: ResolvedContentOptions, diff --git a/packages/eve/src/tracing/content-span-processor.test.ts b/packages/eve/src/tracing/content-span-processor.test.ts index 2da3459d0b..b899ee369f 100644 --- a/packages/eve/src/tracing/content-span-processor.test.ts +++ b/packages/eve/src/tracing/content-span-processor.test.ts @@ -1,35 +1,241 @@ import { describe, expect, it, vi } from "vitest"; +import { + BasicTracerProvider, + type SpanProcessor as OpenTelemetrySpanProcessor, +} from "@opentelemetry/sdk-trace-base"; + +import type { SpanProcessor } from "#compiled/@vercel/otel/index.js"; import { contentFilteringProcessor } from "#tracing/content-span-processor.js"; +function recordingProcessor(): SpanProcessor & { + readonly ended: unknown[]; + readonly started: unknown[]; +} { + const ended: unknown[] = []; + const started: unknown[] = []; + return { + ended, + forceFlush: () => Promise.resolve(), + onEnd: (span) => { + ended.push(span); + }, + onStart: (span) => { + started.push(span); + }, + started, + shutdown: () => Promise.resolve(), + }; +} + +function span(attributes: Record): unknown { + return { + attributes, + spanContext: () => ({ spanId: "span", traceId: "trace" }), + }; +} + describe("contentFilteringProcessor", () => { - it("gives one destination a redacted copy without mutating the shared span", () => { - const onEnd = vi.fn(); - const downstream = { - forceFlush: async () => undefined, - onEnd, - onStart: () => undefined, - shutdown: async () => undefined, - }; + it("forwards the span untouched when the destination declined nothing", () => { + const downstream = recordingProcessor(); + const original = span({ "ai.prompt.messages": "what the user said" }); + + contentFilteringProcessor(downstream, { recordInputs: true, recordOutputs: true }).onEnd( + original as never, + ); + + expect(downstream.ended).toEqual([original]); + }); + + it("forwards a copy without what the destination declined", () => { + const downstream = recordingProcessor(); + + contentFilteringProcessor(downstream, { recordInputs: false, recordOutputs: true }).onEnd( + span({ + "ai.prompt.messages": "what the user said", + "ai.response.text": "what the model said", + }) as never, + ); + + expect((downstream.ended[0] as { attributes: unknown }).attributes).toEqual({ + "ai.response.text": "what the model said", + }); + }); + + it("leaves the original span's attributes in place for the other destinations", () => { + const kept = recordingProcessor(); + const declined = recordingProcessor(); + const original = span({ "ai.prompt.messages": "what the user said" }); + + contentFilteringProcessor(declined, { recordInputs: false, recordOutputs: false }).onEnd( + original as never, + ); + kept.onEnd(original as never); + + expect((declined.ended[0] as { attributes: unknown }).attributes).toEqual({}); + expect((kept.ended[0] as { attributes: unknown }).attributes).toEqual({ + "ai.prompt.messages": "what the user said", + }); + }); + + it("keeps the rest of the span surface reachable on the copy", () => { + const downstream = recordingProcessor(); + + contentFilteringProcessor(downstream, { recordInputs: false, recordOutputs: false }).onEnd( + span({ "ai.prompt.messages": "what the user said" }) as never, + ); + + expect((downstream.ended[0] as { spanContext: () => unknown }).spanContext()).toEqual({ + spanId: "span", + traceId: "trace", + }); + }); + + it("redacts initial attributes before onStart without exposing the original", () => { + const downstream = recordingProcessor(); + const original = span({ + "ai.prompt.messages": "what the user said", + "service.name": "weather", + }); + + contentFilteringProcessor(downstream, { recordInputs: false, recordOutputs: true }).onStart( + original as never, + undefined as never, + ); + + expect(downstream.started[0]).not.toBe(original); + expect((downstream.started[0] as { attributes: unknown }).attributes).toEqual({ + "service.name": "weather", + }); + expect((original as { attributes: unknown }).attributes).toHaveProperty( + "ai.prompt.messages", + "what the user said", + ); + }); + + it("reuses and refreshes one facade from onStart through onEnd", () => { + const downstream = recordingProcessor(); + const original = span({ + "ai.prompt.messages": "what the user said", + "service.name": "weather", + }) as { attributes: Record }; const processor = contentFilteringProcessor(downstream, { recordInputs: false, recordOutputs: true, }); - const span = { - attributes: { - "ai.prompt.messages": "secret", - "ai.response.text": "answer", - "service.name": "weather", + + processor.onStart(original as never, undefined as never); + const retainedAttributes = (downstream.started[0] as { attributes: unknown }).attributes; + original.attributes["ai.response.text"] = "what the model said"; + processor.onEnd(original as never); + + expect(downstream.started[0]).toBe(downstream.ended[0]); + expect((downstream.ended[0] as { attributes: unknown }).attributes).toBe(retainedAttributes); + expect((downstream.started[0] as { attributes: unknown }).attributes).toEqual({ + "ai.response.text": "what the model said", + "service.name": "weather", + }); + }); + + it("keeps SDK methods bound to the original span", () => { + let original: { + attributes: Record; + fluent(): unknown; + setAttribute(key: string, value: unknown): unknown; + spanContext(): unknown; + }; + original = { + attributes: { "ai.prompt.messages": "what the user said" }, + fluent() { + return this; + }, + setAttribute(key, value) { + this.attributes[key] = value; + return this; + }, + spanContext() { + if (this !== original) throw new Error("wrong span receiver"); + return { spanId: "span", traceId: "trace" }; }, - spanContext: () => ({ traceFlags: 1 }), }; + const downstream = recordingProcessor(); + const processor = contentFilteringProcessor(downstream, { + recordInputs: false, + recordOutputs: true, + }); - processor.onEnd(span); + processor.onStart(original as never, undefined as never); - expect(onEnd.mock.calls[0]?.[0].attributes).toStrictEqual({ - "ai.response.text": "answer", - "service.name": "weather", + expect((downstream.started[0] as { spanContext(): unknown }).spanContext()).toEqual({ + spanId: "span", + traceId: "trace", + }); + expect((downstream.started[0] as { fluent(): unknown }).fluent()).toBe(downstream.started[0]); + expect((downstream.started[0] as { valueOf(): unknown }).valueOf()).toBe(downstream.started[0]); + const facade = downstream.started[0] as { + attributes: Record; + setAttribute(key: string, value: unknown): unknown; + }; + expect(facade.setAttribute("service.name", "weather")).toBe(facade); + expect(facade.attributes["service.name"]).toBe("weather"); + }); + + it("continues refreshing after a processor freezes the facade", () => { + const downstream = recordingProcessor(); + const original = span({ "ai.prompt.messages": "what the user said" }) as { + attributes: Record; + }; + const processor = contentFilteringProcessor(downstream, { + recordInputs: false, + recordOutputs: true, }); - expect(span.attributes).toHaveProperty("ai.prompt.messages", "secret"); + + processor.onStart(original as never, undefined as never); + Object.freeze(downstream.started[0]); + original.attributes["ai.response.text"] = "what the model said"; + + expect(() => processor.onEnd(original as never)).not.toThrow(); + expect((downstream.ended[0] as { attributes: unknown }).attributes).toEqual({ + "ai.response.text": "what the model said", + }); + }); + + it("facades a real OpenTelemetry span across both callbacks", () => { + const downstream = recordingProcessor(); + const filtering = contentFilteringProcessor(downstream, { + recordInputs: false, + recordOutputs: true, + }); + const provider = new BasicTracerProvider({ + spanProcessors: [filtering as OpenTelemetrySpanProcessor], + }); + const span = provider.getTracer("test").startSpan("test", { + attributes: { "ai.prompt.messages": "what the user said" }, + }); + + span.setAttribute("ai.response.text", "what the model said"); + span.end(); + + expect(downstream.started[0]).toBe(downstream.ended[0]); + expect((downstream.started[0] as { spanContext(): unknown }).spanContext()).toEqual( + span.spanContext(), + ); + expect((downstream.ended[0] as { attributes: unknown }).attributes).toEqual({ + "ai.response.text": "what the model said", + }); + }); + + it("preserves local trace session release through the wrapper", async () => { + const releaseSession = vi.fn(async () => true); + const downstream: SpanProcessor & { + releaseSession(sessionId: string): Promise; + } = { ...recordingProcessor(), releaseSession }; + const processor = contentFilteringProcessor(downstream, { + recordInputs: false, + recordOutputs: false, + }) as SpanProcessor & { releaseSession(sessionId: string): Promise }; + + await expect(processor.releaseSession("session-1")).resolves.toBe(true); + expect(releaseSession).toHaveBeenCalledExactlyOnceWith("session-1"); }); }); diff --git a/packages/eve/src/tracing/content-span-processor.ts b/packages/eve/src/tracing/content-span-processor.ts index b1c1b05873..3ea8362818 100644 --- a/packages/eve/src/tracing/content-span-processor.ts +++ b/packages/eve/src/tracing/content-span-processor.ts @@ -6,18 +6,51 @@ import { } from "#tracing/content-attributes.js"; import { hasSessionRelease, type LocalTracesProcessor } from "#tracing/local-traces.js"; -/** Gives one destination a redacted copy without mutating the shared span. */ +/** + * Puts one destination's content policy in front of it. + * + * Content is written onto a span if any destination wants it, so the span + * reaching a destination that declined still carries it. This cannot strip the + * attribute in place: that span object is shared with every other processor in + * the pipeline, and editing it would strip the attribute everywhere. So it + * gives the destination a facade whose attributes omit what it declined. + * + * One facade follows the original from start through end. This preserves the + * object identity stateful processors key on without ever exposing the original + * span or its attribute map. Methods stay bound to the original, so private SDK + * state remains reachable without eve knowing that SDK's concrete span shape. + * + * @internal + */ export function contentFilteringProcessor( downstream: SpanProcessor, content: ResolvedContentOptions, ): SpanProcessor { + if (content.recordInputs && content.recordOutputs) return downstream; + + const facades = new WeakMap(); const filtering: SpanProcessor = { forceFlush: () => downstream.forceFlush(), onEnd: (span) => { - downstream.onEnd(redacted(span, content)); + if (typeof span !== "object" || span === null) { + downstream.onEnd(span); + return; + } + + const scoped = facadeFor(span, content, facades); + scoped.refresh(); + try { + downstream.onEnd(scoped.value); + } finally { + facades.delete(span); + } }, onStart: (span, parentContext) => { - downstream.onStart(span, parentContext); + if (typeof span !== "object" || span === null) { + downstream.onStart(span, parentContext); + return; + } + downstream.onStart(facadeFor(span, content, facades).value, parentContext); }, shutdown: () => downstream.shutdown(), }; @@ -30,16 +63,79 @@ export function contentFilteringProcessor( return releasing; } -function redacted(span: unknown, content: ResolvedContentOptions): unknown { - if (typeof span !== "object" || span === null) return span; +interface SpanFacade { + readonly refresh: () => void; + readonly value: object; +} - const attributes = (span as { readonly attributes?: unknown }).attributes; - if (typeof attributes !== "object" || attributes === null) return span; +function facadeFor( + span: object, + content: ResolvedContentOptions, + facades: WeakMap, +): SpanFacade { + const existing = facades.get(span); + if (existing !== undefined) return existing; + + const attributes: Record = {}; + const target = Object.create(Reflect.getPrototypeOf(span)) as Record; + const boundMethods = new Map(); + const refresh = (): void => refreshAttributes(attributes, span, content); + let value: object; + const readOriginal = (property: PropertyKey): unknown => { + const original = Reflect.get(span, property, span) as unknown; + if (typeof original !== "function" || property === "constructor") return original; + + const bound = boundMethods.get(property); + if (bound !== undefined) return bound; + const created = (...args: unknown[]) => { + const result = Reflect.apply(original, span, args) as unknown; + refresh(); + return result === span ? value : result; + }; + boundMethods.set(property, created); + return created; + }; - const kept = withoutDeclinedContent(attributes as Record, content); - if (kept === undefined) return span; + Object.defineProperty(target, "attributes", { + configurable: true, + enumerable: true, + value: attributes, + }); + for (const property of Reflect.ownKeys(span)) { + if (property === "attributes") continue; + const descriptor = Reflect.getOwnPropertyDescriptor(span, property); + Object.defineProperty(target, property, { + configurable: true, + enumerable: descriptor?.enumerable ?? false, + get: () => readOriginal(property), + }); + } - return Object.create(span, { - attributes: { configurable: true, enumerable: true, value: kept }, + value = new Proxy(target, { + get: (facadeTarget, property, receiver) => + Object.hasOwn(facadeTarget, property) + ? Reflect.get(facadeTarget, property, receiver) + : readOriginal(property), }); + const facade = { + refresh, + value, + }; + facade.refresh(); + facades.set(span, facade); + return facade; +} + +function refreshAttributes( + destination: Record, + span: object, + content: ResolvedContentOptions, +): void { + for (const key of Object.keys(destination)) delete destination[key]; + + const source = (span as { readonly attributes?: unknown }).attributes; + if (typeof source !== "object" || source === null) return; + + const kept = withoutDeclinedContent(source as Record, content); + Object.assign(destination, kept ?? source); } diff --git a/packages/eve/src/tracing/local-instrumentation-runtime.ts b/packages/eve/src/tracing/local-instrumentation-runtime.ts index 5c4d8a5e19..c9d21980c2 100644 --- a/packages/eve/src/tracing/local-instrumentation-runtime.ts +++ b/packages/eve/src/tracing/local-instrumentation-runtime.ts @@ -3,7 +3,7 @@ import { type InstrumentationRuntime, } from "#harness/instrumentation-runtime.js"; import { installInstrumentationRuntime } from "#tracing/install-instrumentation-runtime.js"; -import { createLocalTracesProcessor } from "#tracing/local-traces.js"; +import { createLocalTracesProcessor, resolveLocalTracesContent } from "#tracing/local-traces.js"; import { collectOtelPipeline, otel, otelIntegration } from "#tracing/otel-declaration.js"; /** Installs the zero-config local OTel runtime once in an `eve dev` worker. */ @@ -15,24 +15,12 @@ export function installLocalInstrumentationRuntime(input: { const existing = getInstrumentationRuntime(); if (existing !== undefined) return existing; - // The zero-config default expressed with the same values an authored - // `agent/instrumentation/` would declare, so this path exercises them. const spool = createLocalTracesProcessor({ appRoot: input.appRoot }); - const collectedPipeline = collectOtelPipeline([ - otel(), - otelIntegration({ spanProcessors: [spool] }), - ]); - const captureContent = process.env.EVE_TRACES_CONTENT !== "off"; - const collected = { - ...collectedPipeline, - settings: { - ...collectedPipeline.settings, - recordInputs: captureContent, - recordOutputs: captureContent, - }, - }; return installInstrumentationRuntime({ - collected, + collected: collectOtelPipeline([ + otel(), + otelIntegration({ ...resolveLocalTracesContent(), spanProcessors: [spool] }), + ]), frameworkVersion: input.frameworkVersion, providers: [], serviceName: input.serviceName, diff --git a/packages/eve/src/tracing/local-traces.ts b/packages/eve/src/tracing/local-traces.ts index 5878005793..dbbd81ea93 100644 --- a/packages/eve/src/tracing/local-traces.ts +++ b/packages/eve/src/tracing/local-traces.ts @@ -81,7 +81,16 @@ export function createLocalTracesProcessor( }; } -/** Intersects the local destination policy with its environment override. */ +/** + * The local spool's content policy: its options, intersected with + * `EVE_TRACES_CONTENT`. + * + * The variable used to be the process-wide switch. It now applies to this one + * destination, and only ever narrows — `off` still wins where it applies, but + * it no longer decides what a hosted backend beside it receives. + * + * @internal + */ export function resolveLocalTracesContent( options: { readonly recordInputs?: boolean; diff --git a/packages/eve/src/tracing/otel-declaration.test.ts b/packages/eve/src/tracing/otel-declaration.test.ts index 5c677c3047..09a8eed6d3 100644 --- a/packages/eve/src/tracing/otel-declaration.test.ts +++ b/packages/eve/src/tracing/otel-declaration.test.ts @@ -56,6 +56,20 @@ describe("otelIntegration", () => { expect(integration.spanProcessors).toHaveLength(2); expect(integration.spanProcessors[0]).toBe(first); }); + + it("records everything unless told otherwise", () => { + expect(otelIntegration().content).toStrictEqual({ recordInputs: true, recordOutputs: true }); + }); + + // An author's own processor is part of this destination, and the point of + // declining is that nothing under it sees what was said. + it("puts a declined policy in front of every processor, an author's included", () => { + const first = processor(); + const integration = otelIntegration({ recordOutputs: false, spanProcessors: [first] }); + + expect(integration.content).toStrictEqual({ recordInputs: true, recordOutputs: false }); + expect(integration.spanProcessors[0]).not.toBe(first); + }); }); describe("agentRunsIntegration", () => { @@ -123,12 +137,34 @@ describe("collectOtelPipeline", () => { }); expect(collected.settings).toStrictEqual({ functionId: "weather", + // Nothing declared a destination, so nothing asked for content. recordInputs: false, recordOutputs: false, traceChannelRequests: true, }); }); + // Content governs what is written onto the span, which is upstream of every + // destination — so one that wants it is enough, and the ones that declined + // drop it on their own way out. + it("takes content capture as the union across destinations", () => { + const collected = collectOtelPipeline([ + otelIntegration({ recordInputs: false, recordOutputs: false }), + otelIntegration({ recordInputs: false, recordOutputs: true }), + ]); + + expect(collected.settings).toMatchObject({ recordInputs: false, recordOutputs: true }); + }); + + it("writes nothing when every destination declined", () => { + const collected = collectOtelPipeline([ + otelIntegration({ recordInputs: false, recordOutputs: false }), + otelIntegration({ recordInputs: false, recordOutputs: false }), + ]); + + expect(collected.settings).toMatchObject({ recordInputs: false, recordOutputs: false }); + }); + // A process has one tracer provider, so letting the first declaration win // would silently discard the second — the failure this throw exists to stop. it("refuses a second otel() rather than picking one", () => { diff --git a/packages/eve/src/tracing/otel-declaration.ts b/packages/eve/src/tracing/otel-declaration.ts index 18fc5545b3..abfd256140 100644 --- a/packages/eve/src/tracing/otel-declaration.ts +++ b/packages/eve/src/tracing/otel-declaration.ts @@ -51,15 +51,15 @@ export interface OtelOptions { readonly propagators?: readonly PropagatorOrName[]; } -/** Where one `otelIntegration()` sends spans. */ -export interface OtelIntegrationOptions { - /** Merged into the pipeline in declaration order. */ - readonly spanProcessors?: readonly SpanProcessor[]; - /** Wrapped in eve's batching processor and appended after `spanProcessors`. */ - readonly traceExporter?: SpanExporter; -} - -/** What one built-in destination records of the conversation itself. */ +/** + * What one destination records of the conversation itself. + * + * Declining is per destination, not per process: content is written onto the + * span if any destination wants it, and one that declined never exports it. So + * an agent whose every destination declines still never materializes a prompt — + * the union of nothing is nothing — but a local spool and a hosted backend no + * longer have to agree. + */ export interface ContentOptions { /** Record model prompts and tool call inputs. Defaults to `true`. */ readonly recordInputs?: boolean; @@ -67,6 +67,14 @@ export interface ContentOptions { readonly recordOutputs?: boolean; } +/** Where one `otelIntegration()` sends spans, and what it records. */ +export interface OtelIntegrationOptions extends ContentOptions { + /** Merged into the pipeline in declaration order. */ + readonly spanProcessors?: readonly SpanProcessor[]; + /** Wrapped in eve's batching processor and appended after `spanProcessors`. */ + readonly traceExporter?: SpanExporter; +} + const OTEL_DECLARATION = Symbol.for("eve.instrumentation.otel"); const OTEL_INTEGRATION = Symbol.for("eve.instrumentation.otel-integration"); @@ -82,6 +90,7 @@ export interface OtelDeclaration extends InstrumentationProvider { /** One declared destination. A process may have as many as it has files. */ export interface OtelIntegration extends InstrumentationProvider { readonly [OTEL_INTEGRATION]: true; + /** Resolved from `ContentOptions`, so the union does not re-apply defaults. */ readonly content: ResolvedContentOptions; readonly spanProcessors: readonly SpanProcessorOrName[]; } @@ -103,17 +112,30 @@ export function otel(options: OtelOptions = {}): OtelDeclaration { * A `traceExporter` is wrapped in eve's batching processor, which is what makes * the one-line form of a hosted backend enough. Pass `spanProcessors` instead * when the destination needs its own batching, sampling, or filtering. + * + * Declining content wraps every processor here, an author's included: they are + * this destination, and the point of declining is that nothing under it sees + * what was said. */ export function otelIntegration(options: OtelIntegrationOptions = {}): OtelIntegration { + const content: ResolvedContentOptions = { + recordInputs: options.recordInputs !== false, + recordOutputs: options.recordOutputs !== false, + }; const declared = options.spanProcessors ?? []; + const spanProcessors = + options.traceExporter === undefined + ? declared + : [...declared, batchSpanProcessor(options.traceExporter)]; + return { [OTEL_INTEGRATION]: true, [PROVIDER]: true, - content: { recordInputs: true, recordOutputs: true }, + content, spanProcessors: - options.traceExporter === undefined - ? declared - : [...declared, batchSpanProcessor(options.traceExporter)], + content.recordInputs && content.recordOutputs + ? spanProcessors + : spanProcessors.map((processor) => contentFilteringProcessor(processor, content)), }; } @@ -167,8 +189,9 @@ export interface OtelHarnessSettings { readonly functionId?: string; readonly traceChannelRequests: boolean; /** - * What to materialize on spans at all. Each destination independently strips - * anything it declined before export. + * What to write onto a span at all, as opposed to what any one destination + * exports. `agent/instrumentation.ts` sets this directly; a provider + * directory arrives at it as the union of its destinations. */ readonly recordInputs?: boolean; readonly recordOutputs?: boolean; @@ -193,6 +216,10 @@ export interface CollectedOtel { * happened to visit first. With one declaration per file that collision needs * two files both exporting `otel()`, which is the only way to reach it. * + * Content capture is the union of what the destinations asked for, because it + * governs what is written rather than what is exported. Each destination's own + * processors already drop what it declined. + * * @internal */ export function collectOtelPipeline(values: readonly unknown[]): CollectedOtel {