diff --git a/.changeset/construct-channel-instrumentation.md b/.changeset/construct-channel-instrumentation.md new file mode 100644 index 0000000000..7a0e52d539 --- /dev/null +++ b/.changeset/construct-channel-instrumentation.md @@ -0,0 +1,5 @@ +--- +"eve": minor +--- + +Channel audience is now resolved into a delivery decision used to construct harness instrumentation. `tracePolicy` returns an explicit drop or record decision with input/output controls, and audience is no longer exposed through harness lifecycle events or span export policy context. diff --git a/.changeset/vercel-session-id-attribute.md b/.changeset/vercel-session-id-attribute.md new file mode 100644 index 0000000000..86d9ce4566 --- /dev/null +++ b/.changeset/vercel-session-id-attribute.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Agent spans now carry `vercel.session_id` (set to the root session id) when running on Vercel, so traces can be equality-looked-up across all session windows via an indexed column. The attribute is omitted in local `eve dev`. diff --git a/docs/guides/instrumentation.md b/docs/guides/instrumentation.md index 71c26eea64..dd96fa0d69 100644 --- a/docs/guides/instrumentation.md +++ b/docs/guides/instrumentation.md @@ -60,6 +60,8 @@ The third configurable surface, [runtime context events](#runtime-context), atta Built-in messaging channels classify their instrumentation metadata with an `audience`: `public`, `private`, or `unknown`. Slack public channels and Chat SDK workspace-visible threads are public; direct and private conversations are private; platform surfaces without enough visibility evidence remain unknown. +With instrumentation providers enabled, `otel({ tracePolicy })` maps that classification to a decision before the harness runs. eve constructs the harness instrumentation from that decision. Return `{ action: "drop" }` to omit the trace, or `{ action: "record", recordInputs, recordOutputs }` to create it with an explicit content ceiling. The harness receives neither audience nor the decision. + ## Channel delivery traces Instrumentation providers receive `channel.delivery.started` followed by diff --git a/packages/eve/src/context/keys.ts b/packages/eve/src/context/keys.ts index e44d05c394..79e1f78ab2 100644 --- a/packages/eve/src/context/keys.ts +++ b/packages/eve/src/context/keys.ts @@ -27,6 +27,7 @@ import type { SandboxAccess } from "#sandbox/state.js"; import type { RunMode } from "#shared/run-mode.js"; import type { RuntimeModelReference } from "#runtime/agent/bootstrap.js"; import type { PreparedRuntimeDelegationTool } from "#runtime/sessions/turn.js"; +import type { InstrumentationDecision } from "#shared/instrumentation-decision.js"; // Re-export so consumers don't need a direct channel/ import. export type { SessionAuthContext, SessionParent, SessionTurn } from "#channel/types.js"; @@ -89,6 +90,9 @@ export const ActiveChannelDeliveriesKey = new ContextKey( "eve.channelInstrumentation", ); +export const InstrumentationDecisionKey = new ContextKey( + "eve.instrumentationDecision", +); export const ModeKey = new ContextKey("eve.mode"); export const ParentSessionKey = new ContextKey("eve.parentSession"); /** Separate from {@link ParentSessionKey} so it stays out of what extensions read. */ diff --git a/packages/eve/src/execution/delivery-instrumentation.test.ts b/packages/eve/src/execution/delivery-instrumentation.test.ts new file mode 100644 index 0000000000..9efb66d517 --- /dev/null +++ b/packages/eve/src/execution/delivery-instrumentation.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it, vi } from "vitest"; + +import { ContextContainer } from "#context/container.js"; +import { ChannelInstrumentationKey, InstrumentationDecisionKey } from "#context/keys.js"; +import { prepareDeliveryInstrumentation } from "#execution/delivery-instrumentation.js"; +import { createInstrumentationHooks } from "#harness/instrumentation/lifecycle.js"; +import { createInstrumentationRuntime } from "#harness/instrumentation/runtime.js"; +import type { InstrumentationRuntime } from "#harness/instrumentation/runtime.js"; + +function runtime(resolveDecision: InstrumentationRuntime["resolveDecision"]) { + return createInstrumentationRuntime({ + forceFlush: async () => undefined, + hooks: createInstrumentationHooks([]), + otelSettings: { + recordInputs: true, + recordOutputs: true, + traceChannelRequests: false, + }, + resolveDecision, + runInContext: (_operation, execute) => execute(), + runWithTracingSuppressed: (execute) => execute(), + shutdown: async () => undefined, + }); +} + +describe("prepareDeliveryInstrumentation", () => { + it("resolves audience before constructing harness capabilities", () => { + const ctx = new ContextContainer(); + ctx.set(ChannelInstrumentationKey, { + kind: "channel:test", + metadata: { audience: "public" }, + }); + const resolveDecision = vi.fn(() => ({ + action: "record" as const, + recordInputs: false, + recordOutputs: true, + })); + + const constructed = prepareDeliveryInstrumentation({ + ctx, + delivery: { kind: "deliver" }, + instrumentation: runtime(resolveDecision), + rootSessionId: "session-1", + sessionId: "session-1", + }); + + expect(resolveDecision).toHaveBeenCalledWith(expect.objectContaining({ audience: "public" })); + expect(ctx.get(InstrumentationDecisionKey)).toEqual({ + action: "record", + recordInputs: false, + recordOutputs: true, + }); + expect(constructed.harness).not.toHaveProperty("resolveDecision"); + }); + + it("constructs dropped deliveries without provider hooks", () => { + const constructed = prepareDeliveryInstrumentation({ + ctx: new ContextContainer(), + delivery: { kind: "deliver" }, + instrumentation: runtime(vi.fn(() => ({ action: "drop" as const }))), + rootSessionId: "session-1", + sessionId: "session-1", + }); + + expect(constructed.harness?.hooks).toBeUndefined(); + expect(constructed.harness?.telemetryIntegrations).toEqual([]); + }); +}); diff --git a/packages/eve/src/execution/delivery-instrumentation.ts b/packages/eve/src/execution/delivery-instrumentation.ts new file mode 100644 index 0000000000..385b083c4b --- /dev/null +++ b/packages/eve/src/execution/delivery-instrumentation.ts @@ -0,0 +1,47 @@ +import type { ContextContainer } from "#context/container.js"; +import { ChannelInstrumentationKey, InstrumentationDecisionKey } from "#context/keys.js"; +import { + getInstrumentationRuntime, + type ConstructedInstrumentation, + type InstrumentationRuntime, +} from "#harness/instrumentation/runtime.js"; +import { normalizeChannelAudience } from "#shared/channel-audience.js"; + +const UNINSTRUMENTED: ConstructedInstrumentation = { run: (execute) => execute() }; + +/** Resolves audience once, then constructs the only instrumentation visible to the harness. */ +export function prepareDeliveryInstrumentation(input: { + readonly agentName?: string; + readonly ctx: ContextContainer; + readonly delivery?: { readonly kind: string }; + readonly instrumentation?: InstrumentationRuntime; + readonly rootSessionId: string; + readonly sessionId: string; +}): ConstructedInstrumentation { + const instrumentation = input.instrumentation; + if (instrumentation === undefined) return UNINSTRUMENTED; + + let decision = input.ctx.get(InstrumentationDecisionKey); + if (decision === undefined || input.delivery?.kind === "deliver") { + decision = instrumentation.resolveDecision({ + agentName: input.agentName, + audience: normalizeChannelAudience( + input.ctx.get(ChannelInstrumentationKey)?.metadata.audience, + ), + rootSessionId: input.rootSessionId, + sessionId: input.sessionId, + }); + input.ctx.set(InstrumentationDecisionKey, decision); + } + return instrumentation.construct(decision); +} + +export function reconstructInstrumentation( + serializedContext: Record, +): ConstructedInstrumentation { + const instrumentation = getInstrumentationRuntime(); + const decision = serializedContext[InstrumentationDecisionKey.name]; + return instrumentation !== undefined && typeof decision === "object" && decision !== null + ? instrumentation.construct(decision as Parameters[0]) + : UNINSTRUMENTED; +} diff --git a/packages/eve/src/execution/node-step.ts b/packages/eve/src/execution/node-step.ts index eb40d0ceeb..fd77853d88 100644 --- a/packages/eve/src/execution/node-step.ts +++ b/packages/eve/src/execution/node-step.ts @@ -10,8 +10,8 @@ import type { HarnessToolDefinition } from "#harness/execute-tool.js"; import { LOAD_SKILL_TOOL_NAME } from "#runtime/skills/fragment-context.js"; import { createToolLoopHarness } from "#harness/tool-loop.js"; import type { HandleEventFn, HarnessToolMap, StepFn } from "#harness/types.js"; +import type { HarnessInstrumentation } from "#harness/instrumentation/runtime.js"; import { resolveInstalledPackageInfo } from "#internal/application/package.js"; -import { getInstrumentationRuntime } from "#harness/instrumentation/runtime.js"; import { createLogger } from "#internal/logging.js"; import type { RuntimeIdentity } from "#protocol/message.js"; import { UNSPECIFIED_INPUT_SCHEMA } from "#shared/tool-schema.js"; @@ -81,6 +81,7 @@ export interface CreateExecutionNodeStepInput { readonly handleEvent?: HandleEventFn; readonly historyProjector?: HistoryViewProjector; readonly historyView?: PreparedHistoryView; + readonly instrumentation?: HarnessInstrumentation; readonly mode: RunMode; readonly modelResolutionScope: RuntimeModelResolutionScope; readonly node: ResolvedRuntimeAgentNode; @@ -105,7 +106,7 @@ export function createExecutionNodeStep(input: CreateExecutionNodeStepInput): St input.node.turnAgent.dynamicModel, ); const tools = createNodeHarnessTools({ node: input.node }); - const instrumentation = getInstrumentationRuntime(); + const instrumentation = input.instrumentation; const step = createToolLoopHarness({ abortSignal: input.abortSignal, capabilities: input.capabilities, @@ -133,7 +134,7 @@ export function createExecutionNodeStep(input: CreateExecutionNodeStepInput): St try { return await step(session, stepInput); } finally { - await instrumentation.forceFlush(); + await instrumentation.forceFlush?.(); } }; } diff --git a/packages/eve/src/execution/settle-cancelled-turn-step.ts b/packages/eve/src/execution/settle-cancelled-turn-step.ts index 5a0193659e..5bc534a186 100644 --- a/packages/eve/src/execution/settle-cancelled-turn-step.ts +++ b/packages/eve/src/execution/settle-cancelled-turn-step.ts @@ -5,6 +5,7 @@ import { withContextScope } from "#context/run-step.js"; import { deserializeContext, serializeContext } from "#context/serialize.js"; import { ChannelInstrumentationKey } from "#context/keys.js"; import { setChannelContext } from "#execution/channel-context.js"; +import { reconstructInstrumentation } from "#execution/delivery-instrumentation.js"; import { createDurableSessionState, type DurableSessionState, @@ -28,7 +29,6 @@ import { import { abandonRunningAgentTurns } from "#harness/handles/transitions.js"; import { clearPendingRuntimeActionBatch } from "#harness/runtime-actions.js"; import { createInstrumentationHandleEvent } from "#harness/instrumentation/native-events.js"; -import { getInstrumentationRuntime } from "#harness/instrumentation/runtime.js"; import { getTurnUsageState, toUsage } from "#harness/turn-tag-state.js"; import { clearPendingWorkflowInterrupt } from "#harness/workflow-interrupt-state.js"; import { @@ -65,7 +65,7 @@ export async function settleCancelledTurnStep(input: { const adapterCtx = buildAdapterContext(adapter, ctx); const bundle = ctx.require(BundleKey); const effectiveAgent = resolveEffectiveAgentRuntime(bundle, ctx); - const instrumentation = getInstrumentationRuntime(); + const instrumentation = reconstructInstrumentation(input.serializedContext); let session = hydrateDurableSession({ compactionOverrides: { @@ -109,7 +109,7 @@ export async function settleCancelledTurnStep(input: { agentName: bundle.turnAgent.id, channelKind: ctx.get(ChannelInstrumentationKey)?.kind, handleEvent: baseEmit, - hooks: instrumentation?.hooks, + hooks: instrumentation.harness?.hooks, sessionId: session.sessionId, turnId: activeTurnId(emissionState), }) ?? baseEmit; @@ -121,7 +121,7 @@ export async function settleCancelledTurnStep(input: { emissionState = scoped.result; session = scoped.session; } finally { - await instrumentation?.forceFlush(); + await instrumentation.harness?.forceFlush?.(); writer.releaseLock(); } } diff --git a/packages/eve/src/execution/workflow-steps.ts b/packages/eve/src/execution/workflow-steps.ts index ef05c9d37b..d47ff6aa99 100644 --- a/packages/eve/src/execution/workflow-steps.ts +++ b/packages/eve/src/execution/workflow-steps.ts @@ -98,6 +98,7 @@ import { recordSubagentUsageSpans } from "#execution/subagent-usage-span.js"; import { reconcileSessionContinuationToken } from "#execution/reconcile-session-continuation-token.js"; import { hydrateDurableSession, refreshSessionFromTurnAgent } from "#execution/session.js"; import { createExecutionHistoryView } from "#execution/history-view.js"; +import { prepareDeliveryInstrumentation } from "#execution/delivery-instrumentation.js"; import { resolveRuntimeCompiledArtifactsVersionedCacheKey } from "#runtime/cache-key.js"; import { createWorkflowRuntime } from "#execution/workflow-runtime.js"; import { isTaskToolAvailable, TASK_UPDATE_TOOL_NAME } from "#runtime/framework-tools/tasks.js"; @@ -191,6 +192,14 @@ export async function turnStep(rawInput: TurnStepInput): Promise @@ -198,7 +207,7 @@ export async function turnStep(rawInput: TurnStepInput): Promise instrumentChannelDelivery({ ctx, - hooks: instrumentation?.hooks, + hooks: constructedInstrumentation.harness?.hooks, includeTurn: false, outcome: "completed", }), ); - await instrumentation?.forceFlush(); + await constructedInstrumentation.harness?.forceFlush?.(); const rekeyed = reconcileSessionContinuationToken(ctx, initialSession); const nextSerializedContext = serializeContext(ctx); const nextState = @@ -432,6 +441,7 @@ export async function turnStep(rawInput: TurnStepInput): Promise step(modelSession, stepInput)); }; return runHarnessStep(schemaSession, resolved); diff --git a/packages/eve/src/execution/workflow-trace-context.ts b/packages/eve/src/execution/workflow-trace-context.ts index 264d644286..7474b33a9b 100644 --- a/packages/eve/src/execution/workflow-trace-context.ts +++ b/packages/eve/src/execution/workflow-trace-context.ts @@ -1,31 +1,25 @@ import type { ContextContainer } from "#context/container.js"; -import { - ChannelInstrumentationKey, - ParentSessionKey, - ParentTraceContextKey, -} from "#context/keys.js"; +import { ParentSessionKey, ParentTraceContextKey } from "#context/keys.js"; import type { HarnessEmissionState } from "#harness/emission.js"; -import { getInstrumentationRuntime } from "#harness/instrumentation/runtime.js"; +import type { HarnessInstrumentation } from "#harness/instrumentation/runtime.js"; import { resolveParentLineage } from "#harness/parent-lineage.js"; import { prepareTurnTraceContext } from "#harness/prepare-trace-context.js"; import type { HarnessSession } from "#harness/types.js"; import type { RuntimeIdentity, RuntimeTraceContext } from "#protocol/message.js"; import { ChannelKey } from "#runtime/sessions/runtime-context-keys.js"; -import { normalizeChannelAudience } from "#shared/channel-audience.js"; /** Prepares native tracing for workflow-owned preambles emitted outside the tool loop. */ export async function prepareWorkflowPreambleTrace(input: { readonly ctx: ContextContainer; readonly emissionState: HarnessEmissionState; + readonly instrumentation?: HarnessInstrumentation; readonly runtimeIdentity: RuntimeIdentity; readonly session: HarnessSession; }): Promise { const parent = input.ctx.get(ParentSessionKey); - const channel = input.ctx.get(ChannelInstrumentationKey); return await prepareTurnTraceContext({ agentName: input.runtimeIdentity.agentName, - channelAudience: normalizeChannelAudience(channel?.metadata.audience), - instrumentation: getInstrumentationRuntime(), + instrumentation: input.instrumentation, parentLineage: resolveParentLineage(parent, input.ctx.get(ChannelKey)), parentTraceContext: input.ctx.get(ParentTraceContextKey), rootSessionId: parent?.rootSessionId ?? input.session.rootSessionId ?? input.session.sessionId, diff --git a/packages/eve/src/harness/channel-delivery-instrumentation.ts b/packages/eve/src/harness/channel-delivery-instrumentation.ts index dabfe29f60..3ede9df1e6 100644 --- a/packages/eve/src/harness/channel-delivery-instrumentation.ts +++ b/packages/eve/src/harness/channel-delivery-instrumentation.ts @@ -2,7 +2,6 @@ import type { DeliverHookPayload, DeliverPayload } from "#channel/types.js"; import type { AlsContext } from "#context/container.js"; import { ActiveChannelDeliveriesKey, - ChannelInstrumentationKey, ParentTraceContextKey, type ActiveChannelDelivery, } from "#context/keys.js"; @@ -12,7 +11,6 @@ import type { InstrumentationHooks, } from "#harness/instrumentation/lifecycle.js"; import { channelDeliveryIdempotencyKey } from "#harness/instrumentation/lifecycle.js"; -import { normalizeChannelAudience } from "#shared/channel-audience.js"; interface ChannelDeliveryStartInstrumentation { readonly agentName?: string; @@ -70,13 +68,9 @@ export async function instrumentChannelDelivery( if (input.hooks === undefined || input.delivery.deliveryMetadata === undefined) return; const active: ActiveChannelDelivery[] = []; - const channelAudience = normalizeChannelAudience( - input.ctx.get(ChannelInstrumentationKey)?.metadata.audience, - ); for (const metadata of input.delivery.deliveryMetadata) { const payload = input.delivery.payloads[metadata.payloadIndex]; const delivery = { - channelAudience, channelKind: metadata.channelKind, channelName: metadata.channelName, deliveryId: metadata.deliveryId, diff --git a/packages/eve/src/harness/instrumentation/config.test.ts b/packages/eve/src/harness/instrumentation/config.test.ts index f8afd599e1..9c2b9483a9 100644 --- a/packages/eve/src/harness/instrumentation/config.test.ts +++ b/packages/eve/src/harness/instrumentation/config.test.ts @@ -83,7 +83,15 @@ describe("instrumentation-config chunk-isolation regression", () => { { agentName: "test-agent" }, ); - expect(getInstrumentationRuntime()?.otelSettings).toEqual({ + const runtime = getInstrumentationRuntime()!; + expect(runtime.traceChannelRequests).toBe(true); + expect( + runtime.construct({ + action: "record", + recordInputs: true, + recordOutputs: false, + }).harness?.otelSettings, + ).toMatchObject({ functionId: "weather", recordInputs: true, recordOutputs: false, @@ -97,7 +105,15 @@ describe("instrumentation-config chunk-isolation regression", () => { await registerInstrumentationConfig({}, { agentName: "test-agent" }); - expect(getInstrumentationRuntime()?.otelSettings).toEqual({ + const runtime = getInstrumentationRuntime()!; + expect(runtime.traceChannelRequests).toBe(false); + expect( + runtime.construct({ + action: "record", + recordInputs: false, + recordOutputs: false, + }).harness?.otelSettings, + ).toMatchObject({ functionId: undefined, recordInputs: false, recordOutputs: false, diff --git a/packages/eve/src/harness/instrumentation/config.ts b/packages/eve/src/harness/instrumentation/config.ts index c8bf7d8d84..63c6fe1666 100644 --- a/packages/eve/src/harness/instrumentation/config.ts +++ b/packages/eve/src/harness/instrumentation/config.ts @@ -1,5 +1,8 @@ import { createInstrumentationHooks } from "#harness/instrumentation/lifecycle.js"; -import { registerInstrumentationRuntime } from "#harness/instrumentation/runtime.js"; +import { + createInstrumentationRuntime, + registerInstrumentationRuntime, +} from "#harness/instrumentation/runtime.js"; import { createInstrumentationSetupContext } from "#harness/instrumentation/setup-context.js"; import type { InstrumentationDefinition } from "#public/instrumentation/index.js"; @@ -46,18 +49,29 @@ export async function registerInstrumentationConfig( globalContainer[INSTRUMENTATION_CONFIG_GLOBAL_KEY] = config; // This legacy layout leaves `registerOTel` to `setup`, so install only the // runtime projection consumed by the harness. - registerInstrumentationRuntime({ - forceFlush: async () => undefined, - hooks: createInstrumentationHooks([]), - otelSettings: { - functionId: config.functionId, - recordInputs: config.recordInputs === true, - recordOutputs: config.recordOutputs === true, - traceChannelRequests: config.traceChannelRequests === true, - }, - runInContext: (_operation, execute) => execute(), - shutdown: async () => undefined, - }); + const otelSettings = { + functionId: config.functionId, + recordInputs: config.recordInputs === true, + recordOutputs: config.recordOutputs === true, + traceChannelRequests: config.traceChannelRequests === true, + }; + registerInstrumentationRuntime( + createInstrumentationRuntime({ + authoredConfig: config, + forceFlush: async () => undefined, + hooks: createInstrumentationHooks([]), + otelSettings, + resolveDecision: () => ({ + action: "record", + recordInputs: config.recordInputs === true, + recordOutputs: config.recordOutputs === true, + }), + runInContext: (_operation, execute) => execute(), + runWithTracingSuppressed: (execute) => execute(), + shutdown: async () => undefined, + }), + otelSettings, + ); await config.setup?.(createInstrumentationSetupContext(input.agentName)); } diff --git a/packages/eve/src/harness/instrumentation/content.ts b/packages/eve/src/harness/instrumentation/content.ts index 89c7138363..fa6939fa87 100644 --- a/packages/eve/src/harness/instrumentation/content.ts +++ b/packages/eve/src/harness/instrumentation/content.ts @@ -1,31 +1,55 @@ import type { InstrumentationEvent } from "#harness/instrumentation/lifecycle.js"; +import type { InstrumentationDecision } from "#shared/instrumentation-decision.js"; /** Returns an immutable event projection with conversation content removed. */ export function withoutInstrumentationContent(event: InstrumentationEvent): InstrumentationEvent { + return withInstrumentationDecision(event, { + action: "record", + recordInputs: false, + recordOutputs: false, + }); +} + +export function withInstrumentationDecision( + event: InstrumentationEvent, + decision: Extract, +): InstrumentationEvent { switch (event.type) { case "channel.delivery.started": - return Object.freeze({ ...event, input: undefined }); + return decision.recordInputs ? event : Object.freeze({ ...event, input: undefined }); case "action.started": - return Object.freeze({ ...event, input: undefined }); + return decision.recordInputs ? event : Object.freeze({ ...event, input: undefined }); case "action.completed": - return Object.freeze({ ...event, output: Object.freeze({ type: event.output.type }) }); + return decision.recordOutputs + ? event + : Object.freeze({ ...event, output: Object.freeze({ type: event.output.type }) }); case "input.requested": - return Object.freeze({ ...event, request: undefined }); + return decision.recordOutputs ? event : Object.freeze({ ...event, request: undefined }); case "input.resolved": - return Object.freeze({ ...event, error: undefined, response: undefined }); + return decision.recordInputs && decision.recordOutputs + ? event + : Object.freeze({ + ...event, + error: decision.recordOutputs ? event.error : undefined, + response: decision.recordInputs ? event.response : undefined, + }); case "tool.call.started": - return Object.freeze({ ...event, input: undefined }); + return decision.recordInputs ? event : Object.freeze({ ...event, input: undefined }); case "tool.call.completed": - return Object.freeze({ ...event, output: Object.freeze({ type: event.output.type }) }); + return decision.recordOutputs + ? event + : Object.freeze({ ...event, output: Object.freeze({ type: event.output.type }) }); case "model.call.started": - return Object.freeze({ ...event, input: undefined }); + return decision.recordInputs ? event : Object.freeze({ ...event, input: undefined }); case "model.call.completed": - return Object.freeze({ ...event, content: undefined }); + return decision.recordOutputs ? event : Object.freeze({ ...event, content: undefined }); case "step.attempt.metadata": - return Object.freeze({ - ...event, - providerMetadata: structuralProviderMetadata(event.providerMetadata), - }); + return decision.recordOutputs + ? event + : Object.freeze({ + ...event, + providerMetadata: structuralProviderMetadata(event.providerMetadata), + }); case "action.failed": case "model.call.failed": case "session.failed": @@ -33,7 +57,7 @@ export function withoutInstrumentationContent(event: InstrumentationEvent): Inst case "tool.call.failed": case "turn.failed": case "channel.delivery.failed": - return Object.freeze({ ...event, error: undefined }); + return decision.recordOutputs ? event : Object.freeze({ ...event, error: undefined }); default: return event; } diff --git a/packages/eve/src/harness/instrumentation/lifecycle.ts b/packages/eve/src/harness/instrumentation/lifecycle.ts index 9598608c0f..239e64a5a2 100644 --- a/packages/eve/src/harness/instrumentation/lifecycle.ts +++ b/packages/eve/src/harness/instrumentation/lifecycle.ts @@ -1,7 +1,6 @@ import { createInstrumentationDispatcher } from "#harness/instrumentation/dispatch.js"; import type { InstrumentationStateSlot } from "#harness/instrumentation/state.js"; import type { RuntimeTraceContext } from "#protocol/message.js"; -import type { ChannelAudience } from "#shared/channel-audience.js"; /** * Stable eve identity for one actual model attempt. @@ -13,7 +12,6 @@ import type { ChannelAudience } from "#shared/channel-audience.js"; * fire once per attempt. */ export interface InstrumentationAttemptScope { - readonly channelAudience?: ChannelAudience; readonly attemptId: string; readonly attemptIndex: number; readonly functionId?: string; @@ -162,7 +160,6 @@ export function channelDeliveryIdempotencyKey(sessionId: string, deliveryId: str } export interface InstrumentationChannelDeliveryRef { - readonly channelAudience?: ChannelAudience; readonly channelKind: string; readonly channelName: string; readonly deliveryId: string; @@ -284,7 +281,6 @@ export interface InstrumentationSessionStartedEvent { readonly type: "session.started"; readonly agentName?: string; readonly channelKind?: string; - readonly channelAudience?: ChannelAudience; readonly idempotencyKey: string; readonly parentTraceContext?: InstrumentationTraceContext; readonly rootSessionId: string; diff --git a/packages/eve/src/harness/instrumentation/native-events.ts b/packages/eve/src/harness/instrumentation/native-events.ts index 0987c371f1..5dfc62bad1 100644 --- a/packages/eve/src/harness/instrumentation/native-events.ts +++ b/packages/eve/src/harness/instrumentation/native-events.ts @@ -29,12 +29,10 @@ import type { ResolvedInputBatch } from "#harness/input-requests.js"; import { RuntimeActionSettlementTimesKey } from "#harness/runtime-action-settlement-state.js"; import type { HandleEventFn } from "#harness/types.js"; import type { RuntimeActionRequest, RuntimeActionResult } from "#runtime/actions/types.js"; -import type { ChannelAudience } from "#shared/channel-audience.js"; export interface CreateInstrumentationHandleEventInput { readonly agentName?: string; readonly channelKind?: string; - readonly channelAudience?: ChannelAudience; readonly getAttemptScope?: () => InstrumentationAttemptScope | undefined; readonly handleEvent?: HandleEventFn; readonly hooks?: InstrumentationHooks; @@ -294,7 +292,6 @@ function toLifecycleEvent( case "session.started": return { agentName: input.agentName, - channelAudience: input.channelAudience, channelKind: input.channelKind, idempotencyKey: sessionIdempotencyKey(input.sessionId), parentTraceContext: input.parentTraceContext, diff --git a/packages/eve/src/harness/instrumentation/providers.integration.test.ts b/packages/eve/src/harness/instrumentation/providers.integration.test.ts index fc2741f0d9..0deef121bd 100644 --- a/packages/eve/src/harness/instrumentation/providers.integration.test.ts +++ b/packages/eve/src/harness/instrumentation/providers.integration.test.ts @@ -69,7 +69,12 @@ describe("authored instrumentation provider dispatch", () => { expect(order).toEqual(["first:setup", "first:setup-complete", "second:setup"]); const runtime = finalizeInstrumentationProviders({ serviceName: "weather" }); - const publication = runtime.hooks.publish({ + const instrumentation = runtime.construct({ + action: "record", + recordInputs: true, + recordOutputs: true, + }).harness!; + const publication = instrumentation.hooks!.publish({ idempotencyKey: turnIdempotencyKey("session-1", "turn-1"), rootSessionId: "session-1", sequence: 0, diff --git a/packages/eve/src/harness/instrumentation/providers.test.ts b/packages/eve/src/harness/instrumentation/providers.test.ts index 80a4964fb6..0658982c15 100644 --- a/packages/eve/src/harness/instrumentation/providers.test.ts +++ b/packages/eve/src/harness/instrumentation/providers.test.ts @@ -208,7 +208,13 @@ describe("finalizeInstrumentationProviders", () => { await register("rows", defineInstrumentation({ events: { "turn.started": started } })); const runtime = finalizeInstrumentationProviders({ serviceName: "weather-agent" }); - await runtime.hooks.publish(turnStarted); + await runtime + .construct({ + action: "record", + recordInputs: true, + recordOutputs: true, + }) + .harness!.hooks!.publish(turnStarted); expect(started).toHaveBeenCalledOnce(); expect(started.mock.calls[0]?.[0]).toMatchObject({ turnId: "turn-1" }); @@ -221,7 +227,11 @@ describe("finalizeInstrumentationProviders", () => { await register("rows", defineInstrumentation({})); const runtime = finalizeInstrumentationProviders({ serviceName: "weather-agent" }); - const result = await runtime.runInContext( + const result = await runtime.construct({ + action: "record", + recordInputs: true, + recordOutputs: true, + }).harness!.runInContext!( { idempotencyKey: "tool:session-1:turn-1:0:0:call-1:0", scope: { diff --git a/packages/eve/src/harness/instrumentation/runtime.ts b/packages/eve/src/harness/instrumentation/runtime.ts index 188ca6aa80..494fc96106 100644 --- a/packages/eve/src/harness/instrumentation/runtime.ts +++ b/packages/eve/src/harness/instrumentation/runtime.ts @@ -1,3 +1,7 @@ +import type { Telemetry } from "ai"; + +import { getRegisteredTelemetryIntegrations } from "#harness/ai-sdk-telemetry.js"; +import { withInstrumentationDecision } from "#harness/instrumentation/content.js"; import type { InstrumentationContextRunner, InstrumentationHooks, @@ -5,32 +9,127 @@ import type { InstrumentationTraceContext, InstrumentationTurnStartedEvent, } from "#harness/instrumentation/lifecycle.js"; -import type { OtelHarnessSettings, RuntimeContextResolver } from "#tracing/otel-declaration.js"; +import type { InstrumentationDefinition } from "#public/instrumentation/index.js"; +import type { InstrumentationDecision } from "#shared/instrumentation-decision.js"; +import type { + OtelHarnessSettings, + OtelRuntimeSettings, + RuntimeContextResolver, + TraceCaptureContext, +} from "#tracing/otel-declaration.js"; const INSTRUMENTATION_RUNTIME_KEY = Symbol.for("eve.instrumentation-runtime"); +const UPDATE_OTEL_SETTINGS = Symbol.for("eve.instrumentation-runtime.update-otel-settings"); -/** Process-wide runtime consumed by every harness execution surface. */ -export interface InstrumentationRuntime { - readonly forceFlush: () => Promise; - readonly hooks: InstrumentationHooks; +export interface HarnessInstrumentation { + readonly authoredConfig?: InstrumentationDefinition; + readonly forceFlush?: () => Promise; + readonly hooks?: InstrumentationHooks; + readonly otelSettings?: OtelHarnessSettings; readonly prepareSessionTrace?: ( event: InstrumentationSessionStartedEvent, ) => Promise; readonly prepareTurnTrace?: ( event: InstrumentationTurnStartedEvent, ) => Promise; - otelSettings: OtelHarnessSettings | undefined; - /** Provider `runtimeContext` resolvers, collected at install time. */ + readonly runtimeContextResolvers?: readonly RuntimeContextResolver[]; + readonly runInContext?: InstrumentationContextRunner; + readonly telemetryIntegrations?: readonly Telemetry[]; +} + +export interface ConstructedInstrumentation { + readonly harness?: HarnessInstrumentation; + run(execute: () => PromiseLike): PromiseLike; +} + +/** Process runtime that can only resolve and construct scoped instrumentation. */ +export interface InstrumentationRuntime { + readonly [UPDATE_OTEL_SETTINGS]: (settings: OtelRuntimeSettings | undefined) => void; + readonly construct: (decision: InstrumentationDecision) => ConstructedInstrumentation; + readonly forceFlush: () => Promise; + readonly resolveDecision: (context: TraceCaptureContext) => InstrumentationDecision; + readonly shutdown: () => Promise; + readonly traceChannelRequests: boolean; +} + +interface InstrumentationConstructionInput { + readonly authoredConfig?: InstrumentationDefinition; + readonly forceFlush: () => Promise; + readonly hooks: InstrumentationHooks; + readonly otelSettings: OtelRuntimeSettings | undefined; + readonly prepareSessionTrace?: HarnessInstrumentation["prepareSessionTrace"]; + readonly prepareTurnTrace?: HarnessInstrumentation["prepareTurnTrace"]; + readonly resolveDecision: InstrumentationRuntime["resolveDecision"]; readonly runtimeContextResolvers?: readonly RuntimeContextResolver[]; readonly runInContext: InstrumentationContextRunner; + readonly runWithTracingSuppressed: (execute: () => PromiseLike) => PromiseLike; readonly shutdown: () => Promise; } -/** Instrumentation capabilities consumed inside one harness execution. */ -export type HarnessInstrumentation = Pick< - InstrumentationRuntime, - "hooks" | "prepareSessionTrace" | "prepareTurnTrace" | "runInContext" ->; +export function createInstrumentationRuntime( + input: InstrumentationConstructionInput, +): InstrumentationRuntime { + let runtimeSettings = input.otelSettings; + return { + [UPDATE_OTEL_SETTINGS]: (settings) => { + runtimeSettings = settings; + }, + construct: (decision) => construct(input, runtimeSettings, decision), + forceFlush: input.forceFlush, + resolveDecision: input.resolveDecision, + shutdown: input.shutdown, + get traceChannelRequests() { + return runtimeSettings?.traceChannelRequests === true; + }, + }; +} + +function construct( + input: InstrumentationConstructionInput, + runtimeSettings: OtelRuntimeSettings | undefined, + decision: InstrumentationDecision, +): ConstructedInstrumentation { + if (decision.action === "drop") { + return { + harness: { + forceFlush: input.forceFlush, + runInContext: (_operation, execute) => input.runWithTracingSuppressed(execute), + telemetryIntegrations: [], + }, + run: input.runWithTracingSuppressed, + }; + } + + const hooks: InstrumentationHooks = { + capturesContent: + input.hooks.capturesContent && (decision.recordInputs || decision.recordOutputs), + publish: (event) => input.hooks.publish(withInstrumentationDecision(event, decision)), + }; + const otelSettings = + runtimeSettings === undefined + ? undefined + : { + functionId: runtimeSettings.functionId, + recordInputs: runtimeSettings.recordInputs === true && decision.recordInputs, + recordOutputs: runtimeSettings.recordOutputs === true && decision.recordOutputs, + traceChannelRequests: runtimeSettings.traceChannelRequests, + }; + return { + harness: { + authoredConfig: input.authoredConfig, + forceFlush: input.forceFlush, + hooks, + otelSettings, + prepareSessionTrace: input.prepareSessionTrace, + prepareTurnTrace: input.prepareTurnTrace, + runtimeContextResolvers: input.runtimeContextResolvers, + runInContext: input.runInContext, + telemetryIntegrations: + decision.recordInputs && decision.recordOutputs ? getRegisteredTelemetryIntegrations() : [], + }, + run: (execute) => execute(), + }; +} type InstrumentationGlobal = typeof globalThis & { [INSTRUMENTATION_RUNTIME_KEY]?: InstrumentationRuntime; @@ -38,21 +137,19 @@ type InstrumentationGlobal = typeof globalThis & { const globalRuntime = globalThis as InstrumentationGlobal; -/** Registers the process instrumentation runtime before agent execution begins. */ export function registerInstrumentationRuntime( runtime: InstrumentationRuntime, + otelSettings: OtelRuntimeSettings | undefined, ): InstrumentationRuntime { const existing = globalRuntime[INSTRUMENTATION_RUNTIME_KEY]; if (existing !== undefined) { - // A legacy config may reload without taking ownership from the installed runtime. - existing.otelSettings = runtime.otelSettings; + existing[UPDATE_OTEL_SETTINGS](otelSettings); return existing; } globalRuntime[INSTRUMENTATION_RUNTIME_KEY] = runtime; return runtime; } -/** Returns the process instrumentation runtime, when one was installed. */ export function getInstrumentationRuntime(): InstrumentationRuntime | undefined { return globalRuntime[INSTRUMENTATION_RUNTIME_KEY]; } diff --git a/packages/eve/src/harness/prepare-trace-context.ts b/packages/eve/src/harness/prepare-trace-context.ts index d8c1d67358..92873c1e07 100644 --- a/packages/eve/src/harness/prepare-trace-context.ts +++ b/packages/eve/src/harness/prepare-trace-context.ts @@ -6,14 +6,12 @@ import type { } from "#harness/instrumentation/lifecycle.js"; import { sessionIdempotencyKey, turnIdempotencyKey } from "#harness/instrumentation/lifecycle.js"; import type { HarnessInstrumentation } from "#harness/instrumentation/runtime.js"; -import type { ChannelAudience } from "#shared/channel-audience.js"; const log = createLogger("harness.prepare-trace-context"); /** Prepares native session/turn tracing before their durable stream events. */ export async function prepareTurnTraceContext(input: { readonly agentName?: string; - readonly channelAudience?: ChannelAudience; readonly instrumentation?: HarnessInstrumentation; readonly parentLineage?: InstrumentationParentLineage; readonly parentTraceContext?: InstrumentationTraceContext; @@ -30,7 +28,6 @@ export async function prepareTurnTraceContext(input: { try { prepared = await input.instrumentation.prepareSessionTrace({ agentName: input.agentName, - channelAudience: input.channelAudience, idempotencyKey: sessionIdempotencyKey(input.sessionId), parentTraceContext: input.parentTraceContext, rootSessionId: input.rootSessionId, diff --git a/packages/eve/src/harness/tool-loop.test.ts b/packages/eve/src/harness/tool-loop.test.ts index 0e171a9c09..ca53b7f877 100644 --- a/packages/eve/src/harness/tool-loop.test.ts +++ b/packages/eve/src/harness/tool-loop.test.ts @@ -5,6 +5,7 @@ import { type LanguageModelCallEndEvent, type LanguageModel, type ModelMessage, + type Telemetry, ToolLoopAgent, type UserContent, } from "ai"; @@ -39,6 +40,7 @@ import type { ResolvedDynamicInstructionsResolver } from "#runtime/types.js"; import type { DynamicResolveContext } from "#shared/dynamic-tool-definition.js"; import { registerDurableDynamicCallback } from "#shared/durable-dynamic-tool-callbacks.js"; import type { RunMode } from "#shared/run-mode.js"; +import type { OtelHarnessSettings } from "#tracing/otel-declaration.js"; import { compactMessages, shouldCompact } from "#harness/compaction.js"; import { getHarnessEmissionState, isHarnessBetweenTurns } from "#harness/emission.js"; import { @@ -97,7 +99,7 @@ const { registeredOtelIntegration, } = vi.hoisted(() => ({ mockCreateAiSdkHookBridge: vi.fn((..._args: unknown[]) => ({ onStart: vi.fn() })), - mockGetRegisteredTelemetryIntegrations: vi.fn((): unknown[] => []), + mockGetRegisteredTelemetryIntegrations: vi.fn((): Telemetry[] => []), registeredAuthorIntegration: { onStart: vi.fn() }, registeredOtelIntegration: { onStart: vi.fn() }, })); @@ -125,8 +127,17 @@ vi.mock("./instrumentation/runtime.js", () => ({ * Registering an authored config writes both stores, so the tests toggle * telemetry through one call rather than keeping two mocks in step by hand. */ +let declaredOtelSettings: OtelHarnessSettings | undefined; + function declareTelemetry(config: Readonly> | undefined): void { mockGetInstrumentationConfig.mockReturnValue(config); + declaredOtelSettings = + config === undefined + ? undefined + : { + ...config, + traceChannelRequests: config["traceChannelRequests"] === true, + }; mockGetInstrumentationRuntime.mockReturnValue( config === undefined ? undefined @@ -182,6 +193,24 @@ function createTestConfig( emit?: HarnessEmitFn, overrides?: Partial, ): ToolLoopHarnessConfig { + const instrumentation = + overrides?.instrumentation === undefined + ? declaredOtelSettings === undefined + ? undefined + : { + authoredConfig: mockGetInstrumentationConfig() as never, + otelSettings: declaredOtelSettings, + telemetryIntegrations: mockGetRegisteredTelemetryIntegrations(), + } + : { + ...overrides.instrumentation, + authoredConfig: + overrides.instrumentation.authoredConfig ?? (mockGetInstrumentationConfig() as never), + otelSettings: overrides.instrumentation.otelSettings ?? declaredOtelSettings, + telemetryIntegrations: + overrides.instrumentation.telemetryIntegrations ?? + mockGetRegisteredTelemetryIntegrations(), + }; return { handleEvent: emit, mode, @@ -198,6 +227,7 @@ function createTestConfig( ], ]), ...overrides, + instrumentation, }; } @@ -4894,6 +4924,7 @@ describe("createToolLoopHarness", () => { }); const config: ToolLoopHarnessConfig = { instrumentation: { + authoredConfig: mockGetInstrumentationConfig() as never, hooks: createInstrumentationHooks([]), runInContext: (_operation, execute) => execute(), }, @@ -10652,7 +10683,7 @@ describe("createToolLoopHarness", () => { expect(runtimeContext?.["eve.version"]).not.toBe(""); expect(runtimeContext?.["eve.session.id"]).toBe("test-session"); expect(agentCall?.telemetry?.isEnabled).toBe(true); - expect(agentCall?.telemetry?.integrations).toBeUndefined(); + expect(agentCall?.telemetry?.integrations).toEqual([]); }); it("injects one provider-neutral bridge when lifecycle hooks opt in", async () => { diff --git a/packages/eve/src/harness/tool-loop.ts b/packages/eve/src/harness/tool-loop.ts index 5a1e77d5d4..bcb7b4a9fe 100644 --- a/packages/eve/src/harness/tool-loop.ts +++ b/packages/eve/src/harness/tool-loop.ts @@ -55,7 +55,6 @@ import { import { buildDynamicSubagentTools } from "#context/dynamic-subagent-lifecycle.js"; import { PendingSkillAnnouncementKey } from "#context/dynamic-skill-lifecycle.js"; import { toErrorMessage } from "#shared/errors.js"; -import { normalizeChannelAudience } from "#shared/channel-audience.js"; import { createActionResultEvent, createApprovalCandidateEvent, @@ -169,8 +168,6 @@ import { convertStaleResponsesToUserMessage, dropStaleSessionLimitContinuationResponses, } from "#harness/stale-input-responses.js"; -import { getInstrumentationConfig } from "#harness/instrumentation/config.js"; -import { getInstrumentationRuntime } from "#harness/instrumentation/runtime.js"; import type { OtelHarnessSettings } from "#tracing/otel-declaration.js"; import { normalizeModelMessages, @@ -213,10 +210,7 @@ import { TASK_DELIVERY_SETTLED_INSTRUCTION, } from "#tasks/delivery-context.js"; import { extractWorkflowStreamWriteErrorDetails } from "#harness/workflow-stream-error.js"; -import { - ensureOtelIntegration, - getRegisteredTelemetryIntegrations, -} from "#harness/ai-sdk-telemetry.js"; +import { ensureOtelIntegration } from "#harness/ai-sdk-telemetry.js"; import { getAdvertisedTools } from "#harness/advertised-tools.js"; import { createBackgroundToolCallBatch } from "#harness/background-tools.js"; import { @@ -323,8 +317,13 @@ function enrichTelemetry( agentName: string | undefined, runtimeContext?: Readonly>, bridgeIntegration?: Telemetry, + telemetryIntegrations?: readonly Telemetry[], ): TelemetryOptions | undefined { - if (settings === undefined && bridgeIntegration === undefined) { + if ( + settings === undefined && + bridgeIntegration === undefined && + telemetryIntegrations === undefined + ) { return undefined; } @@ -341,9 +340,12 @@ function enrichTelemetry( // Passing integrations replaces the registered ones for this call, so the // bridge has to be composed with them rather than handed over on its own. integrations: - bridgeIntegration === undefined + bridgeIntegration === undefined && telemetryIntegrations === undefined ? undefined - : [bridgeIntegration, ...getRegisteredTelemetryIntegrations()], + : [ + ...(bridgeIntegration === undefined ? [] : [bridgeIntegration]), + ...(telemetryIntegrations ?? []), + ], isEnabled: true, recordInputs: settings?.recordInputs ?? false, recordOutputs: settings?.recordOutputs ?? false, @@ -575,13 +577,12 @@ function buildHarnessToolsWithDynamicSubagents( export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { const baseEmit = config.handleEvent; - const instrumentationRuntime = getInstrumentationRuntime(); - const otelSettings = instrumentationRuntime?.otelSettings; + const otelSettings = config.instrumentation?.otelSettings; // The legacy single-file layout reads its runtime-context resolver from the // authored config object; the provider directory collects resolvers at install // time onto the runtime. Both paths feed buildTelemetryRuntimeContext. - const authoredConfig = getInstrumentationConfig(); - const providerRuntimeContextResolvers = instrumentationRuntime?.runtimeContextResolvers; + const authoredConfig = config.instrumentation?.authoredConfig; + const providerRuntimeContextResolvers = config.instrumentation?.runtimeContextResolvers; if (otelSettings !== undefined) { ensureOtelIntegration(); } @@ -649,7 +650,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { let emissionState = getHarnessEmissionState(session.state); const store = contextStorage.getStore(); - const channelInstrumentation = store?.get(ChannelInstrumentationKey); const parent = store?.get(ParentSessionKey); const channel = store?.get(ChannelKey); const callback = store?.get(SessionCallbackKey); @@ -663,8 +663,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { let activeAttemptScope: InstrumentationAttemptScope | undefined; const emit = createInstrumentationHandleEvent({ agentName: config.runtimeIdentity?.agentName, - channelAudience: normalizeChannelAudience(channelInstrumentation?.metadata.audience), - channelKind: channelInstrumentation?.kind, + channelKind: store?.get(ChannelInstrumentationKey)?.kind, getAttemptScope: () => activeAttemptScope, handleEvent: baseEmit, hooks: config.instrumentation?.hooks, @@ -721,7 +720,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { : undefined; return await prepareTurnTraceContext({ agentName: config.runtimeIdentity?.agentName, - channelAudience: normalizeChannelAudience(channelInstrumentation?.metadata.audience), instrumentation: config.instrumentation, parentLineage, parentTraceContext, @@ -776,7 +774,14 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { resolveModel: config.resolveModel, runtimeIdentity: config.runtimeIdentity, session, - telemetry: enrichTelemetry(otelSettings, agentName) ?? undefined, + telemetry: + enrichTelemetry( + otelSettings, + agentName, + undefined, + undefined, + config.instrumentation?.telemetryIntegrations, + ) ?? undefined, }); session = { @@ -1270,7 +1275,14 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { resolveModel: config.resolveModel, runtimeIdentity: config.runtimeIdentity, session, - telemetry: enrichTelemetry(otelSettings, agentName) ?? undefined, + telemetry: + enrichTelemetry( + otelSettings, + agentName, + undefined, + undefined, + config.instrumentation?.telemetryIntegrations, + ) ?? undefined, }); session = compaction.session; if (compaction.compacted) { @@ -1376,8 +1388,8 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { emissionState, environment, modelInput: { - instructions, - messages: modelMessages, + instructions: otelSettings?.recordInputs === false ? undefined : instructions, + messages: otelSettings?.recordInputs === false ? [] : modelMessages, }, providerResolvers: providerRuntimeContextResolvers, session, @@ -1501,7 +1513,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { : { attemptId: `${session.sessionId}:${instrumentationTurnId}:${emissionState.stepIndex}:${opts.attemptIndex}`, attemptIndex: opts.attemptIndex, - channelAudience: normalizeChannelAudience(channelInstrumentation?.metadata.audience), functionId: otelSettings?.functionId ?? agentName, rootSessionId: parent?.rootSessionId ?? session.sessionId, sessionId: session.sessionId, @@ -1570,6 +1581,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { agentName, telemetryRuntimeContext, bridgeIntegration, + config.instrumentation?.telemetryIntegrations, ), toolApproval: buildToolApproval(modelTools), toolChoice: hasPendingApprovalBatch(session) ? ("none" as const) : undefined, diff --git a/packages/eve/src/internal/nitro/routes/channel-request-instrumentation.ts b/packages/eve/src/internal/nitro/routes/channel-request-instrumentation.ts index 3f9b0431ca..04addc5d78 100644 --- a/packages/eve/src/internal/nitro/routes/channel-request-instrumentation.ts +++ b/packages/eve/src/internal/nitro/routes/channel-request-instrumentation.ts @@ -69,7 +69,7 @@ export async function traceChannelRequest( input: TraceChannelRequestInput, handler: (span: Span | undefined) => Promise, ): Promise { - if (getInstrumentationRuntime()?.otelSettings?.traceChannelRequests !== true) { + if (getInstrumentationRuntime()?.traceChannelRequests !== true) { return await handler(undefined); } diff --git a/packages/eve/src/public/instrumentation/otel.ts b/packages/eve/src/public/instrumentation/otel.ts index 15c43e72bc..11d5cf85b2 100644 --- a/packages/eve/src/public/instrumentation/otel.ts +++ b/packages/eve/src/public/instrumentation/otel.ts @@ -37,6 +37,7 @@ export { type SpanExportPolicy, type SpanExportPredicate, type TraceCaptureContext, + type TraceCaptureDecision, type TraceCapturePolicy, } from "#tracing/otel-declaration.js"; diff --git a/packages/eve/src/shared/instrumentation-decision.ts b/packages/eve/src/shared/instrumentation-decision.ts new file mode 100644 index 0000000000..0b2520498d --- /dev/null +++ b/packages/eve/src/shared/instrumentation-decision.ts @@ -0,0 +1,9 @@ +export type InstrumentationDecision = + | { readonly action: "drop" } + | { + readonly action: "record"; + readonly recordInputs: boolean; + readonly recordOutputs: boolean; + }; + +export const DROP_INSTRUMENTATION: InstrumentationDecision = { action: "drop" }; diff --git a/packages/eve/src/tracing/agent-action-instrumentation.ts b/packages/eve/src/tracing/agent-action-instrumentation.ts index 83de88e60e..8963e9e40f 100644 --- a/packages/eve/src/tracing/agent-action-instrumentation.ts +++ b/packages/eve/src/tracing/agent-action-instrumentation.ts @@ -16,12 +16,11 @@ import type { } from "#harness/instrumentation/lifecycle.js"; import { actionIdempotencyKey, attemptIdempotencyKey } from "#harness/instrumentation/lifecycle.js"; import { contentAttribute } from "#tracing/agent-otel-content.js"; +import { vercelSessionIdAttribute } from "#tracing/agent-otel-attributes.js"; import { setAgentUsage } from "#tracing/agent-otel-usage.js"; import type { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js"; import type { AgentActionTraceState, AgentTraceStateStore } from "#tracing/agent-trace-state.js"; -import { normalizeChannelAudience } from "#shared/channel-audience.js"; import { isSampledTrace } from "#tracing/sampled-trace.js"; -import { withChannelAudience } from "#tracing/channel-audience-context.js"; export interface AgentActionInstrumentation { readonly events: Pick< @@ -45,6 +44,7 @@ export interface AgentActionContext { /** Builds durable `agent.action` spans around eve's runtime dispatch boundary. */ export function createAgentActionInstrumentation(input: { + readonly emitVercelSessionId?: boolean; readonly frameworkVersion: string; readonly idGenerator: AgentSpanIdGenerator; readonly recordInputs: boolean; @@ -55,6 +55,7 @@ export function createAgentActionInstrumentation(input: { readonly stateStore: AgentTraceStateStore; readonly tracer: Tracer; }): AgentActionInstrumentation { + const emitVercelSessionId = input.emitVercelSessionId ?? false; const byAttempt = new Map>(); const onStarted = async (event: InstrumentationActionStartedEvent): Promise => { @@ -65,7 +66,6 @@ export function createAgentActionInstrumentation(input: { const state: AgentActionTraceState = existing ?? { attemptIndex: event.scope.attemptIndex, callId: event.callId, - channelAudience: normalizeChannelAudience(event.scope.channelAudience), inputAttribute: input.recordInputs ? contentAttribute(event.input, false) : undefined, kind: event.kind, name: event.name, @@ -131,6 +131,7 @@ export function createAgentActionInstrumentation(input: { "agent.step.attempt": state.attemptIndex, "agent.step.index": state.stepIndex, "agent.turn.id": state.turnId, + ...vercelSessionIdAttribute(emitVercelSessionId, state.rootSessionId), }, startTime: state.startTimeMs, }, @@ -189,19 +190,13 @@ function actionContext(state: AgentActionTraceState): AgentActionContext { traceId: state.parent.traceId, }; return { - context: withChannelAudience( - trace.setSpan(ROOT_CONTEXT, trace.wrapSpanContext(spanContext)), - state.channelAudience, - ), + context: trace.setSpan(ROOT_CONTEXT, trace.wrapSpanContext(spanContext)), spanContext, }; } function contextFromActionState(state: AgentActionTraceState): Context { - return withChannelAudience( - trace.setSpan(ROOT_CONTEXT, trace.wrapSpanContext({ ...state.parent, isRemote: false })), - state.channelAudience, - ); + return trace.setSpan(ROOT_CONTEXT, trace.wrapSpanContext({ ...state.parent, isRemote: false })); } function recordError(span: Span, error: unknown): void { diff --git a/packages/eve/src/tracing/agent-approval-instrumentation.ts b/packages/eve/src/tracing/agent-approval-instrumentation.ts index cff37245bb..4c243f22cf 100644 --- a/packages/eve/src/tracing/agent-approval-instrumentation.ts +++ b/packages/eve/src/tracing/agent-approval-instrumentation.ts @@ -15,15 +15,14 @@ import type { } from "#harness/instrumentation/lifecycle.js"; import type { JsonValue } from "#shared/json.js"; import { contentAttribute } from "#tracing/agent-otel-content.js"; +import { vercelSessionIdAttribute } from "#tracing/agent-otel-attributes.js"; import type { AgentActionContext } from "#tracing/agent-action-instrumentation.js"; import type { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js"; -import { normalizeChannelAudience, type ChannelAudience } from "#shared/channel-audience.js"; interface AgentApprovalSpanState { readonly actionCallId: string; readonly actionName: string; readonly attemptIndex: number; - readonly channelAudience: ChannelAudience; readonly parent: SpanContext; readonly requestAttribute?: string; readonly requestId: string; @@ -41,6 +40,7 @@ export function createAgentApprovalInstrumentation(input: { turnId: string, callId: string, ) => Promise; + readonly emitVercelSessionId?: boolean; readonly frameworkVersion: string; readonly idGenerator: AgentSpanIdGenerator; readonly recordInputs: boolean; @@ -50,6 +50,7 @@ export function createAgentApprovalInstrumentation(input: { NonNullable, "input.requested" | "input.resolved" > { + const emitVercelSessionId = input.emitVercelSessionId ?? false; const onRequested = async ( event: InstrumentationInputRequestedEvent, ctx: InstrumentationHandlerContext, @@ -65,7 +66,6 @@ export function createAgentApprovalInstrumentation(input: { actionCallId: event.action.callId, actionName: event.action.name, attemptIndex: event.scope.attemptIndex, - channelAudience: normalizeChannelAudience(event.scope.channelAudience), parent: { spanId: parent.spanContext.spanId, traceFlags: parent.spanContext.traceFlags, @@ -110,6 +110,7 @@ export function createAgentApprovalInstrumentation(input: { "agent.step.attempt": state.attemptIndex, "agent.step.index": state.stepIndex, "agent.turn.id": state.turnId, + ...vercelSessionIdAttribute(emitVercelSessionId, state.rootSessionId), }, startTime: state.startTimeMs, }, @@ -161,7 +162,6 @@ function readState(value: unknown): AgentApprovalSpanState | undefined { actionCallId: state["actionCallId"], actionName: state["actionName"], attemptIndex: state["attemptIndex"], - channelAudience: normalizeChannelAudience(state["channelAudience"]), parent: { isRemote: false, spanId: parentRecord["spanId"], diff --git a/packages/eve/src/tracing/agent-channel-delivery-instrumentation.ts b/packages/eve/src/tracing/agent-channel-delivery-instrumentation.ts index d363e9ac50..4ab5a15249 100644 --- a/packages/eve/src/tracing/agent-channel-delivery-instrumentation.ts +++ b/packages/eve/src/tracing/agent-channel-delivery-instrumentation.ts @@ -17,14 +17,13 @@ import type { } from "#harness/instrumentation/lifecycle.js"; import { sessionIdempotencyKey } from "#harness/instrumentation/lifecycle.js"; import type { JsonValue } from "#shared/json.js"; -import { normalizeChannelAudience, type ChannelAudience } from "#shared/channel-audience.js"; import { contentAttribute } from "#tracing/agent-otel-content.js"; +import { vercelSessionIdAttribute } from "#tracing/agent-otel-attributes.js"; import type { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js"; import type { AgentSessionTraceState, AgentTraceStateStore } from "#tracing/agent-trace-state.js"; import { isSampledTrace } from "#tracing/sampled-trace.js"; interface ChannelDeliverySpanState { - readonly channelAudience: ChannelAudience; readonly inputAttribute?: string; readonly parent: SpanContext; readonly requestTraceContext?: SpanContext; @@ -35,6 +34,7 @@ interface ChannelDeliverySpanState { /** Builds durable channel delivery spans around the turn that consumes each request. */ export function createAgentChannelDeliveryInstrumentation(input: { + readonly emitVercelSessionId?: boolean; readonly ensureSessionContext: ( event: InstrumentationSessionStartedEvent, ) => Promise; @@ -50,13 +50,13 @@ export function createAgentChannelDeliveryInstrumentation(input: { | "channel.delivery.failed" | "channel.delivery.started" > { + const emitVercelSessionId = input.emitVercelSessionId ?? false; const onStarted = async ( event: InstrumentationChannelDeliveryStartedEvent, ctx: InstrumentationHandlerContext, ): Promise => { const session = await input.ensureSessionContext({ agentName: event.agentName, - channelAudience: event.delivery.channelAudience, channelKind: event.delivery.channelKind, idempotencyKey: sessionIdempotencyKey(event.sessionId), parentTraceContext: event.parentTraceContext, @@ -67,7 +67,6 @@ export function createAgentChannelDeliveryInstrumentation(input: { if (!isSampledTrace(session.context)) return; const inputAttribute = input.recordInputs ? contentAttribute(event.input, false) : undefined; const state: Record = { - channelAudience: normalizeChannelAudience(event.delivery.channelAudience), parent: { isRemote: session.context.isRemote ?? false, spanId: session.context.spanId, @@ -130,6 +129,7 @@ export function createAgentChannelDeliveryInstrumentation(input: { "agent.session.window": session?.window ?? state.window, "agent.turn.id": event.turnId, "agent.turn.sequence": event.sequence, + ...vercelSessionIdAttribute(emitVercelSessionId, event.rootSessionId), }, kind: SpanKind.CONSUMER, links: @@ -184,7 +184,6 @@ function readState(value: unknown): ChannelDeliverySpanState | undefined { ? value.requestTraceContext : undefined; return { - channelAudience: normalizeChannelAudience(value.channelAudience), inputAttribute: typeof value.inputAttribute === "string" ? value.inputAttribute : undefined, parent: value.parent, requestTraceContext, diff --git a/packages/eve/src/tracing/agent-otel-attributes.ts b/packages/eve/src/tracing/agent-otel-attributes.ts new file mode 100644 index 0000000000..0f24fd7fc8 --- /dev/null +++ b/packages/eve/src/tracing/agent-otel-attributes.ts @@ -0,0 +1,6 @@ +export function vercelSessionIdAttribute( + enabled: boolean, + rootSessionId: string, +): { readonly "vercel.session_id"?: string } { + return enabled ? { "vercel.session_id": rootSessionId } : {}; +} diff --git a/packages/eve/src/tracing/agent-otel-provider.test.ts b/packages/eve/src/tracing/agent-otel-provider.test.ts index 849d4e8497..5b18040b47 100644 --- a/packages/eve/src/tracing/agent-otel-provider.test.ts +++ b/packages/eve/src/tracing/agent-otel-provider.test.ts @@ -36,8 +36,6 @@ import { type InstrumentationTraceContext, type InstrumentationUsage, } from "#harness/instrumentation/lifecycle.js"; -import type { ChannelAudience } from "#shared/channel-audience.js"; -import type { TraceCapturePolicy } from "#tracing/otel-declaration.js"; import { actionIdempotencyKey, attemptIdempotencyKey, @@ -61,7 +59,7 @@ interface TestRuntime { function createRuntime( stateStore: AgentTraceStateStore = new InMemoryAgentTraceStateStore(), - tracePolicy: TraceCapturePolicy | null = () => true, + options: { readonly emitVercelSessionId?: boolean } = {}, ): TestRuntime { const exporter = new InMemorySpanExporter(); const idGenerator = new AgentSpanIdGenerator(); @@ -70,9 +68,8 @@ function createRuntime( spanProcessors: [new SimpleSpanProcessor(exporter)], }); const tracer = provider.getTracer("eve.agent"); - const agentOtelInput: Omit & { - tracePolicy?: TraceCapturePolicy; - } = { + const agentOtelInput: AgentOtelInstrumentationInput = { + emitVercelSessionId: options.emitVercelSessionId, frameworkVersion: "test", idGenerator, recordInputs: true, @@ -80,7 +77,6 @@ function createRuntime( stateStore, tracer, }; - if (tracePolicy !== null) agentOtelInput.tracePolicy = tracePolicy; const agentOtel = createAgentOtelInstrumentation(agentOtelInput); const hooks = createInstrumentationHooks([agentOtel.hook]); return { @@ -98,7 +94,6 @@ async function emitAttempt(input: { readonly actionUsage?: InstrumentationUsage; readonly attemptIndex?: number; readonly attemptError?: Error; - readonly channelAudience?: ChannelAudience; readonly hooks: InstrumentationHooks; readonly parentTraceContext?: InstrumentationTraceContext; readonly runInContext: InstrumentationContextRunner; @@ -116,7 +111,6 @@ async function emitAttempt(input: { const scope: InstrumentationAttemptScope = { attemptId: `${input.sessionId}:${input.turnId}:0:${input.attemptIndex ?? 0}`, attemptIndex: input.attemptIndex ?? 0, - channelAudience: input.channelAudience, functionId: "weather", sessionId: input.sessionId, stepIndex: 0, @@ -286,7 +280,6 @@ async function emitAttempt(input: { } async function publishTurnStarted(input: { - readonly channelAudience?: ChannelAudience; readonly hooks: InstrumentationHooks; readonly parentLineage?: InstrumentationParentLineage; readonly parentTraceContext?: InstrumentationTraceContext; @@ -298,7 +291,6 @@ async function publishTurnStarted(input: { const rootSessionId = input.rootSessionId ?? input.sessionId; await input.hooks.publish({ agentName: "weather", - channelAudience: input.channelAudience, channelKind: "http", idempotencyKey: sessionIdempotencyKey(input.sessionId), parentTraceContext: input.parentTraceContext, @@ -646,7 +638,6 @@ describe("createAgentOtelInstrumentation", () => { const contextWith = vi.spyOn(context, "with"); await context.with(activeContext, () => emitAttempt({ - channelAudience: "public", hooks: runtime.hooks, runInContext: runtime.runInContext, sessionId: "session-1", @@ -672,7 +663,6 @@ describe("createAgentOtelInstrumentation", () => { expect(session.parentSpanContext).toBeUndefined(); expect(session.events.map((event) => event.name)).toEqual(["session.started"]); expect(session.attributes).toMatchObject({ - "agent.channel.audience": "public", "agent.session.id": "session-1", "agent.session.window": 0, "agent.trace.schema.version": 1, @@ -729,62 +719,6 @@ describe("createAgentOtelInstrumentation", () => { }); }); - it.each(["private", "unknown"] as const)( - "does not record %s conversation traces by default", - async (audience) => { - const runtime = createRuntime(new InMemoryAgentTraceStateStore(), null); - - await emitAttempt({ - channelAudience: audience, - hooks: runtime.hooks, - runInContext: runtime.runInContext, - sessionId: `session-${audience}`, - turnId: `turn-${audience}`, - turnSequence: 0, - }); - await runtime.provider.forceFlush(); - - expect(runtime.exporter.getFinishedSpans()).toEqual([]); - }, - ); - - it("applies the private audience gate to an adopted sampled trace", async () => { - const runtime = createRuntime(new InMemoryAgentTraceStateStore(), null); - - await emitAttempt({ - channelAudience: "private", - hooks: runtime.hooks, - parentTraceContext: { - spanId: "a".repeat(16), - traceFlags: 1, - traceId: "b".repeat(32), - }, - runInContext: runtime.runInContext, - sessionId: "session-private", - turnId: "turn-private", - turnSequence: 0, - }); - await runtime.provider.forceFlush(); - - expect(runtime.exporter.getFinishedSpans()).toEqual([]); - }); - - it("allows private tracing when the trace policy opts in", async () => { - const runtime = createRuntime(new InMemoryAgentTraceStateStore(), () => true); - - await emitAttempt({ - channelAudience: "private", - hooks: runtime.hooks, - runInContext: runtime.runInContext, - sessionId: "session-private", - turnId: "turn-private", - turnSequence: 0, - }); - await runtime.provider.forceFlush(); - - expect(byName(runtime.exporter.getFinishedSpans(), "agent.session")).toHaveLength(1); - }); - it("writes merged runtime context onto the step, operation, and chat spans", async () => { const runtime = createRuntime(); @@ -1847,4 +1781,46 @@ describe("createAgentOtelInstrumentation", () => { }); expect(byName(spans, "agent.turn")[0]!.status.code).toBe(SpanStatusCode.UNSET); }); + + describe("emitVercelSessionId", () => { + it("emits vercel.session_id on session, turn, step, and action spans when enabled", async () => { + const runtime = createRuntime(undefined, { emitVercelSessionId: true }); + await emitAttempt({ + hooks: runtime.hooks, + runInContext: runtime.runInContext, + sessionId: "session-1", + turnId: "turn-1", + turnSequence: 0, + }); + await runtime.provider.forceFlush(); + + const spans = runtime.exporter.getFinishedSpans(); + const session = byName(spans, "agent.session")[0]!; + const turn = byName(spans, "agent.turn")[0]!; + const step = byName(spans, "agent.step")[0]!; + const action = byName(spans, "agent.action")[0]!; + + expect(session.attributes["vercel.session_id"]).toBe("session-1"); + expect(turn.attributes["vercel.session_id"]).toBe("session-1"); + expect(step.attributes["vercel.session_id"]).toBe("session-1"); + expect(action.attributes["vercel.session_id"]).toBe("session-1"); + }); + + it("does not emit vercel.session_id by default", async () => { + const runtime = createRuntime(); + await emitAttempt({ + hooks: runtime.hooks, + runInContext: runtime.runInContext, + sessionId: "session-1", + turnId: "turn-1", + turnSequence: 0, + }); + await runtime.provider.forceFlush(); + + const spans = runtime.exporter.getFinishedSpans(); + for (const span of spans) { + expect(span.attributes["vercel.session_id"]).toBeUndefined(); + } + }); + }); }); diff --git a/packages/eve/src/tracing/agent-otel-provider.ts b/packages/eve/src/tracing/agent-otel-provider.ts index 2900345e04..e56f35461c 100644 --- a/packages/eve/src/tracing/agent-otel-provider.ts +++ b/packages/eve/src/tracing/agent-otel-provider.ts @@ -26,10 +26,9 @@ import { createAgentApprovalInstrumentation } from "#tracing/agent-approval-inst import { createAgentChannelDeliveryInstrumentation } from "#tracing/agent-channel-delivery-instrumentation.js"; import { createAgentToolInstrumentation } from "#tracing/agent-tool-instrumentation.js"; import { setAgentUsage } from "#tracing/agent-otel-usage.js"; +import { vercelSessionIdAttribute } from "#tracing/agent-otel-attributes.js"; import { createAgentOtelSessionContext } from "#tracing/agent-otel-session-context.js"; -import type { TraceCapturePolicy } from "#tracing/otel-declaration.js"; import { isSampledTrace } from "#tracing/sampled-trace.js"; -import { withChannelAudience } from "#tracing/channel-audience-context.js"; import { suppressTracing } from "#tracing/suppress-tracing.js"; import type { InstrumentationStepAttemptMetadataEvent, @@ -72,7 +71,14 @@ export interface AgentOtelInstrumentationInput { readonly idGenerator: AgentSpanIdGenerator; readonly stateStore: AgentTraceStateStore; readonly tracer: Tracer; - readonly tracePolicy?: TraceCapturePolicy; + /** + * When true, every agent span also carries `vercel.session_id` set to the + * root session id. The VDP trace ingestion materialized view extracts this + * into the indexed `sessionId` column so Agent Runs can equality-lookup a + * session across all trace windows without scanning the attribute map. + * Emitted only on Vercel (not local `eve dev`). + */ + readonly emitVercelSessionId?: boolean; } /** OTel event definition and its trusted framework context runner. */ @@ -93,6 +99,7 @@ export function createAgentOtelInstrumentation( ): AgentOtelInstrumentation { const recordInputs = input.recordInputs ?? false; const recordOutputs = input.recordOutputs ?? false; + const emitVercelSessionId = input.emitVercelSessionId ?? false; const executionContexts = new WeakMap>(); const attemptScopes = new Map(); // A serverless turn runs inside one `turnStep` "use step" invocation. If @@ -101,6 +108,7 @@ export function createAgentOtelInstrumentation( const steps = new WeakMap(); const modelSpans = new WeakMap>(); const actions = createAgentActionInstrumentation({ + emitVercelSessionId, frameworkVersion: input.frameworkVersion, idGenerator: input.idGenerator, recordInputs, @@ -114,6 +122,7 @@ export function createAgentOtelInstrumentation( }); const approvals = createAgentApprovalInstrumentation({ actionContextFor: actions.contextFor, + emitVercelSessionId, frameworkVersion: input.frameworkVersion, idGenerator: input.idGenerator, recordInputs, @@ -148,10 +157,7 @@ export function createAgentOtelInstrumentation( const onStepStarted = async (event: InstrumentationStepAttemptStartedEvent): Promise => { const turn = await input.stateStore.getTurn(event.scope.sessionId, event.scope.turnId); if (turn === undefined || !isSampledTrace(turn.context)) return; - const turnContext = withChannelAudience( - contextFromSpanContext(turn.context), - event.scope.channelAudience, - ); + const turnContext = contextFromSpanContext(turn.context); const activeSpanContext = trace.getSpan(context.active())?.spanContext(); const stepSpan = input.idGenerator.withSpanId( input.idGenerator.deriveSpanId(attemptIdempotencyKey(event.scope)), @@ -168,6 +174,9 @@ export function createAgentOtelInstrumentation( "agent.step.index": event.scope.stepIndex, "agent.turn.id": event.scope.turnId, "agent.name": event.scope.functionId, + ...(emitVercelSessionId + ? { "vercel.session_id": event.scope.rootSessionId ?? event.scope.sessionId } + : {}), ...runtimeContextAttributes(event.runtimeContext), }, links: @@ -274,6 +283,7 @@ export function createAgentOtelInstrumentation( "agent.session.window": session?.window, "agent.turn.id": event.turnId, "agent.turn.sequence": turn.sequence, + ...vercelSessionIdAttribute(emitVercelSessionId, turn.rootSessionId), }, startTime: turn.startTimeMs, }, @@ -411,6 +421,7 @@ export function createAgentOtelInstrumentation( }; const channelDeliveries = createAgentChannelDeliveryInstrumentation({ + emitVercelSessionId, ensureSessionContext, frameworkVersion: input.frameworkVersion, idGenerator: input.idGenerator, @@ -481,10 +492,7 @@ export function createAgentOtelInstrumentation( operation.scope.turnId, ); if (turn !== undefined) { - parent = withChannelAudience( - contextFromSpanContext(turn.context), - operation.scope.channelAudience, - ); + parent = contextFromSpanContext(turn.context); if (!isSampledTrace(turn.context)) parent = suppressTracing(parent); } } diff --git a/packages/eve/src/tracing/agent-otel-session-context.ts b/packages/eve/src/tracing/agent-otel-session-context.ts index c0f2dbeac1..3e423f337c 100644 --- a/packages/eve/src/tracing/agent-otel-session-context.ts +++ b/packages/eve/src/tracing/agent-otel-session-context.ts @@ -7,9 +7,7 @@ import { type InstrumentationTurnStartedEvent, } from "#harness/instrumentation/lifecycle.js"; import type { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js"; -import { normalizeChannelAudience } from "#shared/channel-audience.js"; -import type { ChannelAudience } from "#shared/channel-audience.js"; -import type { TraceCapturePolicy } from "#tracing/otel-declaration.js"; +import { vercelSessionIdAttribute } from "#tracing/agent-otel-attributes.js"; import { SESSION_WINDOW_TURN_LIMIT, type AgentSessionTraceState, @@ -17,11 +15,11 @@ import { } from "#tracing/agent-trace-state.js"; interface AgentOtelSessionContextInput { + readonly emitVercelSessionId?: boolean; readonly frameworkVersion: string; readonly idGenerator: AgentSpanIdGenerator; readonly stateStore: AgentTraceStateStore; readonly tracer: Tracer; - readonly tracePolicy?: TraceCapturePolicy; } interface AgentOtelSessionContext { @@ -39,32 +37,24 @@ interface AgentOtelSessionContext { export function createAgentOtelSessionContext( input: AgentOtelSessionContextInput, ): AgentOtelSessionContext { + const emitVercelSessionId = input.emitVercelSessionId ?? false; const openSessionWindow = (window: { readonly agentName?: string; - readonly channelAudience: ChannelAudience; readonly index: number; readonly previousTraceId?: string; readonly rootSessionId: string; readonly sessionId: string; }): SpanContext => { - if (!shouldTrace(input.tracePolicy, window)) { - return { - isRemote: false, - spanId: input.idGenerator.deriveSpanId(`session:${window.sessionId}:${window.index}`), - traceFlags: 0, - traceId: input.idGenerator.generateTraceId(), - }; - } const span = input.tracer.startSpan("agent.session", { attributes: { "agent.framework.name": "eve", "agent.framework.version": input.frameworkVersion, - "agent.channel.audience": window.channelAudience, "agent.name": window.agentName, "agent.root.session.id": window.rootSessionId, "agent.session.id": window.sessionId, "agent.session.window": window.index, "agent.trace.schema.version": 1, + ...vercelSessionIdAttribute(emitVercelSessionId, window.rootSessionId), ...(window.previousTraceId === undefined ? {} : { "agent.session.window.previous.trace.id": window.previousTraceId }), @@ -86,26 +76,16 @@ export function createAgentOtelSessionContext( if (state === undefined) { state = { agentName: event.agentName, - channelAudience: normalizeChannelAudience(event.channelAudience), channelKind: event.channelKind, context: event.parentTraceContext === undefined ? openSessionWindow({ agentName: event.agentName, - channelAudience: normalizeChannelAudience(event.channelAudience), index: 0, rootSessionId: event.rootSessionId, sessionId: event.sessionId, }) - : adoptedSpanContext( - event.parentTraceContext, - shouldTrace(input.tracePolicy, { - agentName: event.agentName, - channelAudience: normalizeChannelAudience(event.channelAudience), - rootSessionId: event.rootSessionId, - sessionId: event.sessionId, - }), - ), + : adoptedSpanContext(event.parentTraceContext), rootSessionId: event.rootSessionId, turnsInWindow: 0, window: 0, @@ -125,7 +105,6 @@ export function createAgentOtelSessionContext( ...session, context: openSessionWindow({ agentName: session.agentName, - channelAudience: normalizeChannelAudience(session.channelAudience), index, previousTraceId: session.context.traceId, rootSessionId: session.rootSessionId, @@ -158,7 +137,6 @@ export function createAgentOtelSessionContext( event.sessionId, await ensureSessionContext({ agentName: undefined, - channelAudience: "unknown", channelKind: undefined, idempotencyKey: sessionIdempotencyKey(event.sessionId), parentTraceContext: event.parentTraceContext, @@ -195,29 +173,6 @@ export function createAgentOtelSessionContext( return { ensureSessionContext, prepareSessionTrace, prepareTurnTrace }; } -function shouldTrace( - policy: TraceCapturePolicy | undefined, - trace: { - readonly agentName?: string; - readonly channelAudience: ChannelAudience; - readonly rootSessionId: string; - readonly sessionId: string; - }, -): boolean { - try { - return ( - policy?.({ - agentName: trace.agentName, - audience: trace.channelAudience, - rootSessionId: trace.rootSessionId, - sessionId: trace.sessionId, - }) ?? trace.channelAudience === "public" - ); - } catch { - return false; - } -} - function portableSpanContext(spanContext: SpanContext): InstrumentationTraceContext { return { spanId: spanContext.spanId, @@ -226,11 +181,11 @@ function portableSpanContext(spanContext: SpanContext): InstrumentationTraceCont }; } -function adoptedSpanContext(handed: InstrumentationTraceContext, sampled = true): SpanContext { +function adoptedSpanContext(handed: InstrumentationTraceContext): SpanContext { return { isRemote: "isRemote" in handed && handed.isRemote === true, spanId: handed.spanId, - traceFlags: sampled ? handed.traceFlags : 0, + traceFlags: handed.traceFlags, traceId: handed.traceId, }; } diff --git a/packages/eve/src/tracing/agent-trace-context-store.ts b/packages/eve/src/tracing/agent-trace-context-store.ts index eba434d30e..1041013477 100644 --- a/packages/eve/src/tracing/agent-trace-context-store.ts +++ b/packages/eve/src/tracing/agent-trace-context-store.ts @@ -9,7 +9,6 @@ import type { AgentTraceStateStore, AgentTurnTraceState, } from "#tracing/agent-trace-state.js"; -import { normalizeChannelAudience } from "#shared/channel-audience.js"; interface AgentTraceContextState { readonly actions: Readonly>; @@ -192,7 +191,6 @@ function deserializeState(data: unknown): AgentTraceContextState { if (!isRecord(value) || !isSpanContext(value.context)) return undefined; return { agentName: typeof value.agentName === "string" ? value.agentName : undefined, - channelAudience: normalizeChannelAudience(value.channelAudience), channelKind: typeof value.channelKind === "string" ? value.channelKind : undefined, context: value.context, rootSessionId: typeof value.rootSessionId === "string" ? value.rootSessionId : "", @@ -239,7 +237,6 @@ function deserializeAction(value: unknown): AgentActionTraceState | undefined { return { attemptIndex: value.attemptIndex, callId: value.callId, - channelAudience: normalizeChannelAudience(value.channelAudience), inputAttribute: typeof value.inputAttribute === "string" ? value.inputAttribute : undefined, kind: value.kind, name: value.name, diff --git a/packages/eve/src/tracing/agent-trace-state.ts b/packages/eve/src/tracing/agent-trace-state.ts index 84f0d32cd9..21c79182da 100644 --- a/packages/eve/src/tracing/agent-trace-state.ts +++ b/packages/eve/src/tracing/agent-trace-state.ts @@ -7,13 +7,11 @@ import type { InstrumentationTurnFailedEvent, InstrumentationTurnSettledEvent, } from "#harness/instrumentation/lifecycle.js"; -import type { ChannelAudience } from "#shared/channel-audience.js"; /** Sized so an ordinary session stays one trace and only an outsized one rolls. */ export const SESSION_WINDOW_TURN_LIMIT = 200; export interface AgentSessionTraceState { - readonly channelAudience?: ChannelAudience; readonly agentName?: string; readonly channelKind?: string; readonly context: SpanContext; @@ -38,7 +36,6 @@ export interface AgentTurnTraceState { export interface AgentActionTraceState { readonly attemptIndex: number; readonly callId: string; - readonly channelAudience?: ChannelAudience; readonly inputAttribute?: string; readonly kind: InstrumentationActionKind; readonly name: string; diff --git a/packages/eve/src/tracing/channel-audience-context.ts b/packages/eve/src/tracing/channel-audience-context.ts deleted file mode 100644 index a8884c3bd8..0000000000 --- a/packages/eve/src/tracing/channel-audience-context.ts +++ /dev/null @@ -1,16 +0,0 @@ -import { createContextKey, type Context } from "#compiled/@opentelemetry/api/index.js"; -import { normalizeChannelAudience, type ChannelAudience } from "#shared/channel-audience.js"; - -const CHANNEL_AUDIENCE_KEY = createContextKey("eve.channel.audience"); - -export function channelAudienceFromContext(context: unknown): ChannelAudience { - if (typeof context !== "object" || context === null) return "unknown"; - const getValue = Reflect.get(context, "getValue"); - return typeof getValue === "function" - ? normalizeChannelAudience(Reflect.apply(getValue, context, [CHANNEL_AUDIENCE_KEY])) - : "unknown"; -} - -export function withChannelAudience(context: Context, audience: unknown): Context { - return context.setValue(CHANNEL_AUDIENCE_KEY, normalizeChannelAudience(audience)); -} diff --git a/packages/eve/src/tracing/content-span-processor.test.ts b/packages/eve/src/tracing/content-span-processor.test.ts index 26dd66535e..4f214436e0 100644 --- a/packages/eve/src/tracing/content-span-processor.test.ts +++ b/packages/eve/src/tracing/content-span-processor.test.ts @@ -242,26 +242,25 @@ describe("contentFilteringProcessor", () => { }); it.each([ - ["public", true], - ["private", false], - ["unknown", false], - ] as const)("retains content for the %s audience: %s", (audience, retained) => { + ["visible", true], + ["hidden", false], + ] as const)("retains content when the span is %s: %s", (visibility, retained) => { const downstream = recordingProcessor(); contentFilteringProcessor( downstream, composeSpanExportPolicies( - redactSpanInputs(({ audience }) => audience !== "public"), - redactSpanOutputs(({ audience }) => audience !== "public"), + redactSpanInputs(({ attributes }) => attributes["visibility"] !== "visible"), + redactSpanOutputs(({ attributes }) => attributes["visibility"] !== "visible"), ), ).onEnd( span({ - "agent.channel.audience": audience, + visibility, "ai.prompt.messages": "input", "ai.response.text": "output", }) as never, ); - const expected: Record = { "agent.channel.audience": audience }; + const expected: Record = { visibility }; if (retained) { expected["ai.prompt.messages"] = "input"; expected["ai.response.text"] = "output"; @@ -271,27 +270,6 @@ describe("contentFilteringProcessor", () => { ); }); - it("fails closed when audience attributes disagree", () => { - const downstream = recordingProcessor(); - contentFilteringProcessor( - downstream, - composeSpanExportPolicies( - redactSpanInputs(({ audience }) => audience !== "public"), - redactSpanOutputs(({ audience }) => audience !== "public"), - ), - ).onEnd( - span({ - "agent.channel.audience": "public", - "ai.prompt.messages": "private", - "ai.settings.context.eve.channel.audience": "private", - }) as never, - ); - - expect( - (downstream.ended[0] as { attributes: Record }).attributes, - ).not.toHaveProperty("ai.prompt.messages"); - }); - it("can drop an individual span", () => { const downstream = recordingProcessor(); contentFilteringProcessor(downstream, { span: ({ name }) => name !== "private-work" }).onEnd({ diff --git a/packages/eve/src/tracing/content-span-processor.ts b/packages/eve/src/tracing/content-span-processor.ts index 563162c8c7..774081e57c 100644 --- a/packages/eve/src/tracing/content-span-processor.ts +++ b/packages/eve/src/tracing/content-span-processor.ts @@ -5,15 +5,12 @@ import { type ResolvedContentOptions, } from "#tracing/content-attributes.js"; import { hasSessionRelease, type LocalTracesProcessor } from "#tracing/local-traces.js"; -import { normalizeChannelAudience } from "#shared/channel-audience.js"; -import type { ChannelAudience } from "#shared/channel-audience.js"; import { contentRedactionForSpan, spanExportPolicyStages, type SpanExportContext, type SpanExportPolicy, } from "#tracing/span-export-policy.js"; -import { channelAudienceFromContext } from "#tracing/channel-audience-context.js"; /** * Puts one destination's content policy in front of it. @@ -74,12 +71,7 @@ function policyFilteringProcessor( if (typeof span !== "object" || span === null) { return; } - const scoped = facadeFor( - span, - exportPolicy, - facades, - channelAudienceFromContext(parentContext), - ); + const scoped = facadeFor(span, exportPolicy, facades); if (!scoped.exported) { dropped.add(span); return; @@ -108,12 +100,11 @@ function facadeFor( span: object, exportPolicy: SpanExportPolicy, facades: WeakMap, - inheritedAudience?: ChannelAudience, ): SpanFacade { const existing = facades.get(span); if (existing !== undefined) return existing; - const context = spanExportContext(span, inheritedAudience); + const context = spanExportContext(span); const effectiveContent = contentForSpan(context, exportPolicy); const attributes: Record = {}; const events: unknown[] = []; @@ -212,26 +203,12 @@ function refreshAttributes( } } -function spanExportContext( - span: object, - inheritedAudience: ChannelAudience = "unknown", -): SpanExportContext { +function spanExportContext(span: object): SpanExportContext { const attributes = (span as { readonly attributes?: unknown }).attributes; const record = typeof attributes === "object" && attributes !== null ? (attributes as Readonly>) : {}; - const candidates = [ - record["agent.channel.audience"], - record["ai.settings.context.eve.channel.audience"], - ].filter((value) => value !== undefined); - const normalized = candidates.map(normalizeChannelAudience); - const audience = - normalized.length > 0 && normalized.every((value) => value === normalized[0]) - ? normalized[0]! - : normalized.length > 0 - ? "unknown" - : inheritedAudience; const spanContext = (span as { readonly spanContext?: () => unknown }).spanContext?.(); const ids = typeof spanContext === "object" && spanContext !== null @@ -240,7 +217,6 @@ function spanExportContext( const name = (span as { readonly name?: unknown }).name; return { attributes: record, - audience, name: typeof name === "string" ? name : "", spanId: typeof ids["spanId"] === "string" ? ids["spanId"] : "", traceId: typeof ids["traceId"] === "string" ? ids["traceId"] : "", diff --git a/packages/eve/src/tracing/install-instrumentation-runtime.test.ts b/packages/eve/src/tracing/install-instrumentation-runtime.test.ts index d56ad362a1..8f6ebf88db 100644 --- a/packages/eve/src/tracing/install-instrumentation-runtime.test.ts +++ b/packages/eve/src/tracing/install-instrumentation-runtime.test.ts @@ -69,7 +69,14 @@ describe("installInstrumentationRuntime", () => { expect(forceFlush).toHaveBeenCalledOnce(); expect(providerFlush).toHaveBeenCalledOnce(); - expect(runtime.otelSettings).toEqual({ + expect(runtime.traceChannelRequests).toBe(false); + expect( + runtime.construct({ + action: "record", + recordInputs: true, + recordOutputs: true, + }).harness?.otelSettings, + ).toMatchObject({ functionId: undefined, recordInputs: true, recordOutputs: true, @@ -97,9 +104,14 @@ describe("installInstrumentationRuntime", () => { serviceName: "weather", }); const idempotencyKey = turnIdempotencyKey("session-1", "turn-1"); + const instrumentation = runtime.construct({ + action: "record", + recordInputs: true, + recordOutputs: true, + }).harness!; await contextStorage.run(new ContextContainer(), async () => { - await runtime.hooks.publish({ + await instrumentation.hooks!.publish({ idempotencyKey, rootSessionId: "session-1", sequence: 0, @@ -107,7 +119,7 @@ describe("installInstrumentationRuntime", () => { turnId: "turn-1", type: "turn.started", }); - await runtime.hooks.publish({ + await instrumentation.hooks!.publish({ idempotencyKey, sessionId: "session-1", turnId: "turn-1", @@ -118,4 +130,23 @@ describe("installInstrumentationRuntime", () => { expect(internalTerminalState).toHaveBeenCalledExactlyOnceWith("framework"); expect(authoredTerminalState).toHaveBeenCalledExactlyOnceWith("authored"); }); + + it("resolves audience before constructing harness instrumentation", () => { + const runtime = installInstrumentationRuntime({ + collected: collectOtelPipeline([otelIntegration()]), + frameworkVersion: "test", + providers: [], + serviceName: "weather", + }); + const context = { rootSessionId: "session-1", sessionId: "session-1" }; + + expect(runtime.resolveDecision({ ...context, audience: "public" })).toEqual({ + action: "record", + recordInputs: true, + recordOutputs: true, + }); + expect(runtime.resolveDecision({ ...context, audience: "private" })).toEqual({ + action: "drop", + }); + }); }); diff --git a/packages/eve/src/tracing/install-instrumentation-runtime.ts b/packages/eve/src/tracing/install-instrumentation-runtime.ts index 7c78c5e3f9..e91a05735c 100644 --- a/packages/eve/src/tracing/install-instrumentation-runtime.ts +++ b/packages/eve/src/tracing/install-instrumentation-runtime.ts @@ -1,4 +1,4 @@ -import { trace } from "#compiled/@opentelemetry/api/index.js"; +import { context, trace } from "#compiled/@opentelemetry/api/index.js"; import type { SpanProcessor } from "#compiled/@vercel/otel/index.js"; import { @@ -6,14 +6,19 @@ import { type InstrumentationProviderDefinition, } from "#harness/instrumentation/lifecycle.js"; import { + createInstrumentationRuntime, registerInstrumentationRuntime, + type HarnessInstrumentation, type InstrumentationRuntime, } from "#harness/instrumentation/runtime.js"; import { createLogger, formatError } from "#internal/logging.js"; +import { isEveDevEnvironment } from "#internal/application/dev-environment.js"; import { ContextAgentTraceStateStore } from "#tracing/agent-trace-context-store.js"; import { createAgentOtelInstrumentation } from "#tracing/agent-otel-provider.js"; import { hasSessionRelease, type LocalTracesProcessor } from "#tracing/local-traces.js"; import type { CollectedOtel, RuntimeContextResolver } from "#tracing/otel-declaration.js"; +import { DROP_INSTRUMENTATION } from "#shared/instrumentation-decision.js"; +import { suppressTracing } from "#tracing/suppress-tracing.js"; import { registerOtelPipeline, type RegisteredOtelPipeline } from "#tracing/otel-registration.js"; const log = createLogger("tracing.install-instrumentation-runtime"); @@ -38,9 +43,10 @@ export function installInstrumentationRuntime(input: { const serialBefore: InstrumentationProviderDefinition[] = []; const serialAfter: InstrumentationProviderDefinition[] = []; let otelRuntime: RegisteredOtelPipeline | undefined; - let prepareSessionTrace: InstrumentationRuntime["prepareSessionTrace"]; - let prepareTurnTrace: InstrumentationRuntime["prepareTurnTrace"]; - let runInContext: InstrumentationRuntime["runInContext"] = (_operation, execute) => execute(); + let prepareSessionTrace: HarnessInstrumentation["prepareSessionTrace"]; + let prepareTurnTrace: HarnessInstrumentation["prepareTurnTrace"]; + let runInContext: NonNullable = (_operation, execute) => + execute(); if (input.collected.declared) { otelRuntime = registerOtelPipeline({ @@ -48,13 +54,13 @@ export function installInstrumentationRuntime(input: { serviceName: input.serviceName, }); const agentOtel = createAgentOtelInstrumentation({ + emitVercelSessionId: process.env.VERCEL_ENV !== undefined && !isEveDevEnvironment(), frameworkVersion: input.frameworkVersion, idGenerator: otelRuntime.idGenerator, recordInputs: input.collected.settings.recordInputs, recordOutputs: input.collected.settings.recordOutputs, stateStore: new ContextAgentTraceStateStore(), tracer: trace.getTracer("eve.agent", input.frameworkVersion), - tracePolicy: input.collected.settings.tracePolicy, }); // The span must exist before authored providers observe the lifecycle event. serialBefore.push({ ...agentOtel.hook, stateNamespace: "internal:otel" }); @@ -69,31 +75,53 @@ export function installInstrumentationRuntime(input: { } const allProviders = [...serialBefore, ...input.providers, ...serialAfter]; + const hooks = createInstrumentationHooks({ + parallel: input.providers, + serialAfter, + serialBefore, + }); + const otelSettings = input.collected.declared ? input.collected.settings : undefined; let shutdown: Promise | undefined; - return registerInstrumentationRuntime({ - forceFlush: () => - settleAll([ - ...(otelRuntime === undefined ? [] : [otelRuntime.forceFlush]), - ...allProviders.map((provider) => () => provider.flush?.()), - ]), - hooks: createInstrumentationHooks({ - parallel: input.providers, - serialAfter, - serialBefore, + return registerInstrumentationRuntime( + createInstrumentationRuntime({ + forceFlush: () => + settleAll([ + ...(otelRuntime === undefined ? [] : [otelRuntime.forceFlush]), + ...allProviders.map((provider) => () => provider.flush?.()), + ]), + hooks, + otelSettings, + prepareSessionTrace, + prepareTurnTrace, + resolveDecision: (traceContext) => { + if (!input.collected.declared) { + return { action: "record", recordInputs: true, recordOutputs: true }; + } + try { + return ( + input.collected.settings.tracePolicy?.(traceContext) ?? + (traceContext.audience === "public" + ? { action: "record", recordInputs: true, recordOutputs: true } + : DROP_INSTRUMENTATION) + ); + } catch { + return DROP_INSTRUMENTATION; + } + }, + runtimeContextResolvers: input.runtimeContextResolvers, + runInContext, + runWithTracingSuppressed: (execute) => + context.with(suppressTracing(context.active()), execute), + shutdown: () => { + shutdown ??= settleAll([ + ...(otelRuntime === undefined ? [] : [otelRuntime.shutdown]), + ...allProviders.map((provider) => () => provider.shutdown?.()), + ]); + return shutdown; + }, }), - otelSettings: input.collected.declared ? input.collected.settings : undefined, - prepareSessionTrace, - prepareTurnTrace, - runtimeContextResolvers: input.runtimeContextResolvers, - runInContext, - shutdown: () => { - shutdown ??= settleAll([ - ...(otelRuntime === undefined ? [] : [otelRuntime.shutdown]), - ...allProviders.map((provider) => () => provider.shutdown?.()), - ]); - return shutdown; - }, - }); + otelSettings, + ); } function isSpanProcessor(processor: SpanProcessor | "auto"): processor is SpanProcessor { diff --git a/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts b/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts index bd6c42626e..ba75c71d7c 100644 --- a/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts +++ b/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts @@ -43,11 +43,21 @@ describe("local instrumentation runtime", () => { const authoredTracer = ( require("@opentelemetry/api") as typeof import("@opentelemetry/api") ).trace.getTracer("test-user"); - const runtime = installLocalInstrumentationRuntime({ + const installedRuntime = installLocalInstrumentationRuntime({ appRoot, frameworkVersion: "test", serviceName: "test-agent", }); + const runtime = installedRuntime.construct({ + action: "record", + recordInputs: true, + recordOutputs: true, + }).harness!; + const hooks = runtime.hooks; + const runInContext = runtime.runInContext; + if (hooks === undefined || runInContext === undefined) { + throw new Error("Expected constructed local instrumentation."); + } const scope: InstrumentationAttemptScope = { attemptId: "session-1:turn-1:0:0", attemptIndex: 0, @@ -61,14 +71,14 @@ describe("local instrumentation runtime", () => { const activeContext = runtimeTrace.setSpan(COMPILED_ROOT_CONTEXT, delivery); const exerciseRuntime = async () => { - await runtime.hooks.publish({ + await hooks.publish({ agentName: "weather", idempotencyKey: sessionIdempotencyKey("session-1"), rootSessionId: "session-1", sessionId: "session-1", type: "session.started", }); - await runtime.hooks.publish({ + await hooks.publish({ idempotencyKey: turnIdempotencyKey("session-1", "turn-1"), rootSessionId: "session-1", sequence: 0, @@ -76,7 +86,7 @@ describe("local instrumentation runtime", () => { turnId: "turn-1", type: "turn.started", }); - const bridge = createAiSdkHookBridge(scope, runtime.hooks, runtime.runInContext); + const bridge = createAiSdkHookBridge(scope, hooks, runInContext); Reflect.apply(bridge.onStart!, bridge, [ { callId: "call-1", @@ -115,7 +125,7 @@ describe("local instrumentation runtime", () => { }, ]); const actionKey = actionIdempotencyKey("session-1", "turn-1", "tool-1"); - await runtime.hooks.publish({ + await hooks.publish({ callId: "tool-1", idempotencyKey: actionKey, input: {}, @@ -149,26 +159,26 @@ describe("local instrumentation runtime", () => { toolOutput: { output: { temperature: 72 }, type: "tool-result" }, }, ]); - await runtime.hooks.publish({ + await hooks.publish({ idempotencyKey: actionKey, outcome: "completed", output: { output: { temperature: 72 }, type: "result" }, scope, type: "action.completed", }); - await runtime.hooks.publish({ + await hooks.publish({ idempotencyKey: attemptIdempotencyKey(scope), scope, type: "step.attempt.completed", }); - await runtime.hooks.publish({ + await hooks.publish({ idempotencyKey: turnIdempotencyKey("session-1", "turn-1"), sessionId: "session-1", turnId: "turn-1", type: "turn.completed", }); // Settling the turn emits the turn span with the pre-allocated id. - await runtime.hooks.publish({ + await hooks.publish({ idempotencyKey: sessionIdempotencyKey("session-1"), sessionId: "session-1", turnId: "turn-1", @@ -179,7 +189,7 @@ describe("local instrumentation runtime", () => { contextStorage.run(new ContextContainer(), exerciseRuntime), ); delivery.end(); - await runtime.forceFlush(); + await runtime.forceFlush?.(); const traceRoot = join(appRoot, ".eve", "traces", "v1"); const [traceId] = await readdir(traceRoot); diff --git a/packages/eve/src/tracing/local-instrumentation-runtime.ts b/packages/eve/src/tracing/local-instrumentation-runtime.ts index 905547fb1a..9d55de4246 100644 --- a/packages/eve/src/tracing/local-instrumentation-runtime.ts +++ b/packages/eve/src/tracing/local-instrumentation-runtime.ts @@ -13,7 +13,9 @@ import { /** Zero-config local tracing keeps unclassified HTTP/TUI sessions observable. @internal */ export const localTracePolicy: TraceCapturePolicy = ({ audience }) => - audience === "public" || audience === "unknown"; + audience === "public" || audience === "unknown" + ? { action: "record", recordInputs: true, recordOutputs: true } + : { action: "drop" }; /** Installs the zero-config local OTel runtime once in an `eve dev` worker. */ export function installLocalInstrumentationRuntime(input: { diff --git a/packages/eve/src/tracing/local-traces.test.ts b/packages/eve/src/tracing/local-traces.test.ts index e0bd1e850b..d9d684f1f8 100644 --- a/packages/eve/src/tracing/local-traces.test.ts +++ b/packages/eve/src/tracing/local-traces.test.ts @@ -101,16 +101,16 @@ describe("resolveLocalTracesContent", () => { describe("localTracePolicy", () => { it.each([ - ["public", true], - ["unknown", true], - ["private", false], - ] as const)("accepts the %s audience: %s", (audience, accepted) => { + ["public", "record"], + ["unknown", "record"], + ["private", "drop"], + ] as const)("maps the %s audience to %s", (audience, action) => { expect( localTracePolicy({ audience, rootSessionId: "session-1", sessionId: "session-1", }), - ).toBe(accepted); + ).toMatchObject({ action }); }); }); diff --git a/packages/eve/src/tracing/otel-declaration.test.ts b/packages/eve/src/tracing/otel-declaration.test.ts index 57173f035d..10c836310c 100644 --- a/packages/eve/src/tracing/otel-declaration.test.ts +++ b/packages/eve/src/tracing/otel-declaration.test.ts @@ -116,7 +116,7 @@ describe("managed export policy", () => { let visibleAttributes: Readonly> | undefined; const integration = managedOtelIntegration({ exportPolicy: composeSpanExportPolicies( - redactSpanInputs(({ audience }) => audience !== "public"), + redactSpanInputs(({ attributes }) => attributes["visibility"] !== "public"), { span: ({ attributes }) => { visibleAttributes = attributes; @@ -131,12 +131,12 @@ describe("managed export policy", () => { if (spanProcessor === undefined || spanProcessor === "auto") throw new Error("Expected policy"); spanProcessor.onEnd( testSpan({ - "agent.channel.audience": "private", + visibility: "private", "ai.prompt.messages": "private input", }), ); - expect(visibleAttributes).toEqual({ "agent.channel.audience": "private" }); + expect(visibleAttributes).toEqual({ visibility: "private" }); }); it("applies deprecated content switches before the configured export policy", () => { diff --git a/packages/eve/src/tracing/otel-declaration.ts b/packages/eve/src/tracing/otel-declaration.ts index e14ff8b229..5b7451f337 100644 --- a/packages/eve/src/tracing/otel-declaration.ts +++ b/packages/eve/src/tracing/otel-declaration.ts @@ -9,6 +9,7 @@ import type { import { PROVIDER, type InstrumentationProvider } from "#public/instrumentation/provider.js"; import type { InstrumentationRuntimeContextInput } from "#public/instrumentation/index.js"; import type { JsonObject } from "#shared/json.js"; +import type { InstrumentationDecision } from "#shared/instrumentation-decision.js"; import { batchSpanProcessor } from "#tracing/batch-span-processor.js"; import type { ResolvedContentOptions } from "#tracing/content-attributes.js"; import { contentFilteringProcessor } from "#tracing/content-span-processor.js"; @@ -111,7 +112,9 @@ export interface TraceCaptureContext { readonly sessionId: string; } -export type TraceCapturePolicy = (trace: TraceCaptureContext) => boolean; +export type TraceCaptureDecision = InstrumentationDecision; + +export type TraceCapturePolicy = (trace: TraceCaptureContext) => TraceCaptureDecision; /** Where one `otelIntegration()` sends spans. */ export interface OtelIntegrationOptions extends ContentOptions { @@ -276,12 +279,16 @@ export interface OtelPipeline { export interface OtelHarnessSettings { readonly functionId?: string; readonly traceChannelRequests: boolean; - readonly tracePolicy?: TraceCapturePolicy; /** Legacy `defineInstrumentation()` capture settings. Provider destinations capture fully. */ readonly recordInputs?: boolean; readonly recordOutputs?: boolean; } +/** Process settings retained outside the harness until channel delivery. @internal */ +export interface OtelRuntimeSettings extends OtelHarnessSettings { + readonly tracePolicy?: TraceCapturePolicy; +} + /** @internal */ export type RuntimeContextResolver = ( input: InstrumentationRuntimeContextInput, @@ -296,7 +303,7 @@ export interface CollectedOtel { readonly declared: boolean; readonly pipeline: OtelPipeline; readonly runtimeContextResolvers: readonly RuntimeContextResolver[]; - readonly settings: OtelHarnessSettings; + readonly settings: OtelRuntimeSettings; } /** @@ -337,7 +344,7 @@ export function collectOtelPipeline(values: readonly unknown[]): CollectedOtel { } const options = declaration?.options ?? {}; - const settings: OtelHarnessSettings = { + const settings: OtelRuntimeSettings = { functionId: options.functionId, recordInputs: capturesContent, recordOutputs: capturesContent, diff --git a/packages/eve/src/tracing/span-export-policy.ts b/packages/eve/src/tracing/span-export-policy.ts index 116008179f..343b6556ee 100644 --- a/packages/eve/src/tracing/span-export-policy.ts +++ b/packages/eve/src/tracing/span-export-policy.ts @@ -1,8 +1,5 @@ -import type { ChannelAudience } from "#shared/channel-audience.js"; - export interface SpanExportContext { readonly attributes: Readonly>; - readonly audience: ChannelAudience; readonly name: string; readonly spanId: string; readonly traceId: string; diff --git a/research/channel-audience-content-policy.md b/research/channel-audience-content-policy.md index d88d730dcc..faf5648fc3 100644 --- a/research/channel-audience-content-policy.md +++ b/research/channel-audience-content-policy.md @@ -22,7 +22,7 @@ type ChannelAudience = "public" | "private" | "unknown"; The field is optional for authored channels. Eve normalizes absent, malformed, and unsupported values to `unknown`. Built-in channel metadata interfaces require the field and classify only from platform evidence already captured during dispatch; ambiguous and proactive destinations remain `unknown` rather than performing observability-only network requests. -The normalized audience is persisted with session trace state and exported as `agent.channel.audience` only on each `agent.session` window. Durable Eve state and an internal OpenTelemetry context key make the same value available to descendant export policies without duplicating a public attribute onto every span. Local subagents inherit the parent audience. Remote agents classify their receiving channel independently rather than trusting opaque metadata across deployment boundaries. +At channel delivery, eve evaluates the process-wide trace policy and persists only its generic decision for that turn. eve constructs the harness hooks, telemetry, and trace context from that decision. The harness, lifecycle providers, trace state, and export policies never receive the audience classification or branch on the decision. ## Public tracing API @@ -38,7 +38,15 @@ interface TraceCaptureContext { readonly sessionId: string; } -type TraceCapturePolicy = (trace: TraceCaptureContext) => boolean; +type TraceCaptureDecision = + | { readonly action: "drop" } + | { + readonly action: "record"; + readonly recordInputs: boolean; + readonly recordOutputs: boolean; + }; + +type TraceCapturePolicy = (trace: TraceCaptureContext) => TraceCaptureDecision; interface OtelOptions { // Other process-wide OTel settings are unchanged. @@ -53,7 +61,6 @@ Agent Runs and local traces expose the managed export policy: ```ts interface SpanExportContext { readonly attributes: Readonly>; - readonly audience: ChannelAudience; readonly name: string; readonly spanId: string; readonly traceId: string; @@ -106,20 +113,19 @@ For example, this admits public and private conversations at the head gate while ```ts // agent/instrumentation/otel.ts export default otel({ - tracePolicy: ({ audience }) => audience === "public" || audience === "private", + tracePolicy: ({ audience }) => + audience === "public" + ? { action: "record", recordInputs: true, recordOutputs: true } + : { action: "drop" }, }); // agent/instrumentation/agent-runs.ts export default agentRuns({ - exportPolicy: composeSpanExportPolicies( - redactSpanInputs(({ audience }) => audience !== "public"), - redactSpanOutputs(({ audience }) => audience !== "public"), - { - span: ({ name }) => name !== "internal.cache.refresh", - attribute: ({ key }) => - key === "user.email" ? { action: "replace", value: "[redacted]" } : { action: "keep" }, - }, - ), + exportPolicy: composeSpanExportPolicies({ + span: ({ name }) => name !== "internal.cache.refresh", + attribute: ({ key }) => + key === "user.email" ? { action: "replace", value: "[redacted]" } : { action: "keep" }, + }), }); ``` @@ -128,7 +134,10 @@ export default agentRuns({ The default authored and production head policy is equivalent to: ```ts -({ audience }) => audience === "public"; +({ audience }) => + audience === "public" + ? { action: "record", recordInputs: true, recordOutputs: true } + : { action: "drop" }; ``` | Audience | Trace created by default | Content when admitted by a custom trace policy | @@ -140,21 +149,22 @@ The default authored and production head policy is equivalent to: The default policy for local tracing for `eve dev` is equivalent to: ```ts -({ audience }) => audience === "public" || audience === "unknown"; +({ audience }) => + audience === "public" || audience === "unknown" + ? { action: "record", recordInputs: true, recordOutputs: true } + : { action: "drop" }; ``` This keeps unclassified local HTTP/TUI sessions observable while still rejecting channels classified as `private`. The runtime order is: -1. Derive and normalize the channel audience. -2. Evaluate the process-wide `tracePolicy` before creating `agent.session`. -3. For accepted traces, capture complete Eve and AI SDK spans. +1. Derive and normalize the channel audience at delivery. +2. Evaluate `tracePolicy`, persist its decision, and construct the harness instrumentation. +3. Capture only the inputs and outputs admitted by that constructed capability. 4. Run each managed destination's composed export policies in declaration order. Custom integrations run their declared span processors. 5. Hand the resulting facade to that destination's processors or exporter. -There is no implicit content redaction after a custom trace policy admits an audience. Redaction occurs only when the export pipeline includes `redactSpanInputs()` or `redactSpanOutputs()` (or when a retained compatibility option explicitly requests the equivalent redaction). - Policies fail closed at their boundary: a throwing trace policy rejects the trace, a throwing span policy drops the span, a throwing attribute policy drops the attribute, and a throwing content-redaction predicate redacts that content direction. Missing, malformed, or conflicting audience evidence normalizes to `unknown`. ## Compatibility