diff --git a/.changeset/instrumentation-action-tracing.md b/.changeset/instrumentation-action-tracing.md new file mode 100644 index 0000000000..262f323425 --- /dev/null +++ b/.changeset/instrumentation-action-tracing.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Persist action trace identity across workers and reconstruct backdated `agent.action` spans when durable actions settle. diff --git a/packages/eve/src/tracing/agent-action-instrumentation.ts b/packages/eve/src/tracing/agent-action-instrumentation.ts new file mode 100644 index 0000000000..c2296787c5 --- /dev/null +++ b/packages/eve/src/tracing/agent-action-instrumentation.ts @@ -0,0 +1,155 @@ +import { + ROOT_CONTEXT, + SpanStatusCode, + type Context, + type Span, + type SpanContext, + type Tracer, + trace, +} from "#compiled/@opentelemetry/api/index.js"; + +import type { + InstrumentationActionStartedEvent, + InstrumentationActionTerminalEvent, + InstrumentationProviderDefinition, +} from "#harness/instrumentation-lifecycle.js"; +import { actionIdempotencyKey } from "#harness/instrumentation-lifecycle.js"; +import { contentAttribute } from "#tracing/agent-otel-content.js"; +import type { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js"; +import type { AgentActionTraceState, AgentTraceStateStore } from "#tracing/agent-trace-state.js"; + +export interface AgentActionInstrumentation { + readonly events: Pick< + NonNullable, + "action.completed" | "action.failed" | "action.started" + >; + deleteForSession(sessionId: string): void | PromiseLike; + deleteForTurn(sessionId: string, turnId: string): void | PromiseLike; + contextFor(sessionId: string, turnId: string, callId: string): Promise; +} + +/** Builds durable `agent.action` spans around eve's runtime dispatch boundary. */ +export function createAgentActionInstrumentation(input: { + readonly frameworkVersion: string; + readonly idGenerator: AgentSpanIdGenerator; + readonly recordInputs: boolean; + readonly recordOutputs: boolean; + readonly resolveParent: ( + event: InstrumentationActionStartedEvent, + ) => { readonly context: Context; readonly spanContext: SpanContext } | undefined; + readonly stateStore: AgentTraceStateStore; + readonly tracer: Tracer; +}): AgentActionInstrumentation { + const onStarted = async (event: InstrumentationActionStartedEvent): Promise => { + const parent = input.resolveParent(event); + if (parent === undefined) return; + + const existing = await input.stateStore.getAction(event.idempotencyKey); + const state: AgentActionTraceState = existing ?? { + attemptIndex: event.scope.attemptIndex, + callId: event.callId, + inputAttribute: input.recordInputs ? contentAttribute(event.input, false) : undefined, + kind: event.kind, + name: event.name, + parent: { + spanId: parent.spanContext.spanId, + traceFlags: parent.spanContext.traceFlags, + traceId: parent.spanContext.traceId, + }, + rootSessionId: event.scope.rootSessionId ?? event.scope.sessionId, + sessionId: event.scope.sessionId, + spanId: input.idGenerator.allocateSpanId(), + startTimeMs: Date.now(), + stepIndex: event.scope.stepIndex, + turnId: event.scope.turnId, + }; + await input.stateStore.setAction(event.idempotencyKey, state); + }; + + const onTerminal = async (event: InstrumentationActionTerminalEvent): Promise => { + const state = await input.stateStore.getAction(event.idempotencyKey); + if (state === undefined) return; + try { + const span = startSpan(state); + if (event.type === "action.failed") { + recordError(span, event.error); + } else if (event.output.type === "error") { + recordError(span, event.output.error); + } else if (input.recordOutputs) { + const result = contentAttribute(event.output.output, false); + if (result !== undefined) span.setAttribute("gen_ai.tool.call.result", result); + } + span.end(); + } finally { + await input.stateStore.deleteAction(event.idempotencyKey); + } + }; + + const startSpan = (state: AgentActionTraceState): Span => { + const span = input.idGenerator.withSpanId(state.spanId, () => + input.tracer.startSpan( + "agent.action", + { + attributes: { + "agent.action.call_id": state.callId, + "agent.action.kind": state.kind, + "agent.action.name": state.name, + "agent.framework.name": "eve", + "agent.framework.version": input.frameworkVersion, + "agent.root.session.id": state.rootSessionId, + "agent.session.id": state.sessionId, + "agent.step.attempt": state.attemptIndex, + "agent.step.index": state.stepIndex, + "agent.turn.id": state.turnId, + }, + startTime: state.startTimeMs, + }, + contextFromActionState(state), + ), + ); + if (state.inputAttribute !== undefined) { + span.setAttribute("gen_ai.tool.call.arguments", state.inputAttribute); + } + return span; + }; + + return { + async contextFor(sessionId, turnId, callId) { + const directKey = actionIdempotencyKey(sessionId, turnId, callId); + const direct = await input.stateStore.getAction(directKey); + if (direct !== undefined) return actionContext(direct); + const state = await input.stateStore.findAction(sessionId, callId); + return state === undefined ? undefined : actionContext(state); + }, + deleteForSession: (sessionId) => input.stateStore.deleteActions(sessionId), + deleteForTurn: (sessionId, turnId) => input.stateStore.deleteActions(sessionId, turnId), + events: { + "action.completed": onTerminal, + "action.failed": onTerminal, + "action.started": onStarted, + }, + }; +} + +function actionContext(state: AgentActionTraceState): Context { + return trace.setSpan( + ROOT_CONTEXT, + trace.wrapSpanContext({ + isRemote: false, + spanId: state.spanId, + traceFlags: state.parent.traceFlags, + traceId: state.parent.traceId, + }), + ); +} + +function contextFromActionState(state: AgentActionTraceState): Context { + return trace.setSpan(ROOT_CONTEXT, trace.wrapSpanContext({ ...state.parent, isRemote: false })); +} + +function recordError(span: Span, error: unknown): void { + if (error instanceof Error) { + span.recordException(error); + span.setStatus({ code: SpanStatusCode.ERROR, message: error.message }); + } else span.setStatus({ code: SpanStatusCode.ERROR }); +} diff --git a/packages/eve/src/tracing/agent-otel-provider.test.ts b/packages/eve/src/tracing/agent-otel-provider.test.ts index 11d3fba36d..6ccbc96bd6 100644 --- a/packages/eve/src/tracing/agent-otel-provider.test.ts +++ b/packages/eve/src/tracing/agent-otel-provider.test.ts @@ -7,24 +7,32 @@ import { } from "@opentelemetry/sdk-trace-base"; import { describe, expect, it } from "vitest"; +import { ContextContainer, contextStorage } from "#context/container.js"; +import { deserializeContext, serializeContext } from "#context/serialize.js"; import { createAiSdkHookBridge } from "#harness/ai-sdk-hook-bridge.js"; import { createAgentOtelInstrumentation } from "#tracing/agent-otel-provider.js"; import { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js"; +import { ContextAgentTraceStateStore } from "#tracing/agent-trace-context-store.js"; import { + type AgentTraceStateStore, InMemoryAgentTraceStateStore, SESSION_WINDOW_TURN_LIMIT, } from "#tracing/agent-trace-state.js"; import { - attemptIdempotencyKey, createInstrumentationHooks, - sessionIdempotencyKey, - turnIdempotencyKey, + type InstrumentationActionKind, type InstrumentationAttemptScope, type InstrumentationContextRunner, type InstrumentationHooks, type InstrumentationParentLineage, type InstrumentationTraceContext, } from "#harness/instrumentation-lifecycle.js"; +import { + actionIdempotencyKey, + attemptIdempotencyKey, + sessionIdempotencyKey, + turnIdempotencyKey, +} from "#harness/instrumentation-lifecycle.js"; interface TestRuntime { readonly exporter: InMemorySpanExporter; @@ -33,7 +41,9 @@ interface TestRuntime { readonly runInContext: InstrumentationContextRunner; } -function createRuntime(stateStore = new InMemoryAgentTraceStateStore()): TestRuntime { +function createRuntime( + stateStore: AgentTraceStateStore = new InMemoryAgentTraceStateStore(), +): TestRuntime { const exporter = new InMemorySpanExporter(); const idGenerator = new AgentSpanIdGenerator(); const provider = new BasicTracerProvider({ @@ -55,6 +65,7 @@ async function emitAttempt(input: { readonly hooks: InstrumentationHooks; readonly runInContext: InstrumentationContextRunner; readonly providerMetadata?: Readonly>; + readonly actionKind?: InstrumentationActionKind; readonly sessionId: string; readonly skipModelTerminal?: boolean; readonly skipToolTerminal?: boolean; @@ -137,6 +148,16 @@ async function emitAttempt(input: { }, ]); } + const actionKey = actionIdempotencyKey(input.sessionId, input.turnId, "tool-1"); + await input.hooks.publish({ + callId: "tool-1", + idempotencyKey: actionKey, + input: { secret: "value" }, + kind: input.actionKind ?? "tool-call", + name: "weather", + scope, + type: "action.started", + }); await Reflect.apply(bridge.onToolExecutionStart!, bridge, [ { callId: "call-1", @@ -161,6 +182,15 @@ async function emitAttempt(input: { : { error: input.toolError, type: "tool-error" }, }, ]); + await input.hooks.publish({ + idempotencyKey: actionKey, + output: + input.toolError === undefined + ? { output: { temperature: 72 }, type: "result" } + : { error: input.toolError, type: "error" }, + scope, + type: "action.completed", + }); } if (input.providerMetadata !== undefined) { @@ -267,6 +297,7 @@ describe("createAgentOtelInstrumentation", () => { const step = byName(spans, "agent.step")[0]!; const operation = byName(spans, "ai.streamText")[0]!; const model = byName(spans, "ai.streamText.doStream")[0]!; + const action = byName(spans, "agent.action")[0]!; const tool = byName(spans, "ai.toolCall")[0]!; const session = byName(spans, "agent.session")[0]!; @@ -280,7 +311,8 @@ describe("createAgentOtelInstrumentation", () => { expect(step.parentSpanContext?.spanId).toBe(turn.spanContext().spanId); expect(operation.parentSpanContext?.spanId).toBe(step.spanContext().spanId); expect(model.parentSpanContext?.spanId).toBe(operation.spanContext().spanId); - expect(tool.parentSpanContext?.spanId).toBe(step.spanContext().spanId); + expect(action.parentSpanContext?.spanId).toBe(step.spanContext().spanId); + expect(tool.parentSpanContext?.spanId).toBe(action.spanContext().spanId); expect(new Set(spans.map((span) => span.spanContext().traceId))).toHaveLength(1); expect(turn.events.map((event) => event.name)).toEqual([ "turn.started", @@ -302,9 +334,108 @@ describe("createAgentOtelInstrumentation", () => { "gen_ai.usage.cache_creation.input_tokens": 2, "gen_ai.usage.cache_read.input_tokens": 4, }); + expect(action.attributes).toMatchObject({ + "agent.action.kind": "tool-call", + "agent.action.name": "weather", + "agent.framework.name": "eve", + "agent.root.session.id": "session-1", + }); }); - it("ends model and tool spans still open when the step attempt terminates", async () => { + it("reconstructs a durable action span in a replacement worker", async () => { + const first = createRuntime(new ContextAgentTraceStateStore()); + const context = new ContextContainer(); + const scope: InstrumentationAttemptScope = { + attemptId: "session-1:turn-1:0:0", + attemptIndex: 0, + sessionId: "session-1", + stepIndex: 0, + turnId: "turn-1", + }; + const actionKey = actionIdempotencyKey(scope.sessionId, scope.turnId, "tool-1"); + const toolKey = `tool:${scope.attemptId}:tool-1:0`; + + await contextStorage.run(context, async () => { + await publishTurnStarted({ + hooks: first.hooks, + sessionId: scope.sessionId, + turnId: scope.turnId, + turnSequence: 0, + }); + await first.hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + operation: { modelId: "model", operationId: "ai.streamText", provider: "test" }, + scope, + type: "step.attempt.started", + }); + await first.hooks.publish({ + callId: "tool-1", + idempotencyKey: actionKey, + input: { secret: "value" }, + kind: "tool-call", + name: "weather", + scope, + type: "action.started", + }); + await first.hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + scope, + type: "step.attempt.completed", + }); + }); + await first.provider.forceFlush(); + const firstSpans = first.exporter.getFinishedSpans(); + const step = byName(firstSpans, "agent.step")[0]!; + + await new Promise((resolve) => setTimeout(resolve, 2)); + const restored = await deserializeContext(await serializeContext(context)); + const replacement = createRuntime(new ContextAgentTraceStateStore()); + const replacementScope = { + ...scope, + attemptId: "session-1:turn-2:0:0", + turnId: "turn-2", + }; + await contextStorage.run(restored, async () => { + // Approval-resumed tools can execute before the replacement AI SDK emits + // a new step start. The persisted action context is still their parent. + await replacement.hooks.publish({ + callId: "tool-1", + idempotencyKey: toolKey, + input: {}, + scope: replacementScope, + toolName: "weather", + type: "tool.call.started", + }); + await replacement.hooks.publish({ + idempotencyKey: toolKey, + output: { output: "ok", type: "result" }, + scope: replacementScope, + type: "tool.call.completed", + }); + await replacement.hooks.publish({ + idempotencyKey: actionKey, + output: { output: { temperature: 72 }, type: "result" }, + scope, + type: "action.completed", + }); + }); + await replacement.provider.forceFlush(); + + const replacementSpans = replacement.exporter.getFinishedSpans(); + const action = byName(replacementSpans, "agent.action")[0]!; + const tool = byName(replacementSpans, "ai.toolCall")[0]!; + expect(action.spanContext().spanId).toBe(tool.parentSpanContext?.spanId); + expect(action.parentSpanContext?.spanId).toBe(step.spanContext().spanId); + expect(action.attributes).toMatchObject({ + "agent.action.kind": "tool-call", + "agent.action.name": "weather", + "gen_ai.tool.call.arguments": expect.stringContaining("secret"), + "gen_ai.tool.call.result": expect.stringContaining("temperature"), + }); + expect(nanos(action.duration)).toBeGreaterThan(0n); + }); + + it("ends SDK spans but leaves durable actions for their own terminal", async () => { const runtime = createRuntime(); await emitAttempt({ hooks: runtime.hooks, @@ -324,6 +455,25 @@ describe("createAgentOtelInstrumentation", () => { expect(byName(spans, "agent.step")).toHaveLength(1); }); + it("labels a subagent action by its kind, not as a plain tool", async () => { + const runtime = createRuntime(); + await emitAttempt({ + hooks: runtime.hooks, + actionKind: "subagent-call", + runInContext: runtime.runInContext, + sessionId: "session-1", + turnId: "turn-1", + turnSequence: 0, + }); + await runtime.provider.forceFlush(); + + const action = runtime.exporter.getFinishedSpans().find((span) => span.name === "agent.action"); + expect(action?.attributes).toMatchObject({ + "agent.action.kind": "subagent-call", + "agent.action.name": "weather", + }); + }); + it("captures model and tool inputs/outputs on the operation spans", async () => { const runtime = createRuntime(); await emitAttempt({ @@ -358,6 +508,10 @@ describe("createAgentOtelInstrumentation", () => { ); expect(tool.attributes["gen_ai.tool.call.arguments"]).toBe('{"secret":"value"}'); expect(tool.attributes["gen_ai.tool.call.result"]).toBe('{"temperature":72}'); + // Runtime action spans carry content for dispatches that have no SDK tool boundary. + const action = byName(spans, "agent.action")[0]!; + expect(action.attributes["gen_ai.tool.call.arguments"]).toContain("secret"); + expect(action.attributes["gen_ai.tool.call.result"]).toContain("temperature"); }); it("truncates long conversations from the front, keeping valid JSON and recent messages", async () => { @@ -575,8 +729,10 @@ describe("createAgentOtelInstrumentation", () => { const replacementSpans = replacementRuntime.exporter.getFinishedSpans(); const turn = byName(replacementSpans, "agent.turn")[0]!; const step = byName(replacementSpans, "agent.step")[0]!; + const action = byName(replacementSpans, "agent.action")[0]!; expect(step.parentSpanContext?.spanId).toBe(turn.spanContext().spanId); + expect(action.parentSpanContext?.spanId).toBe(step.spanContext().spanId); expect(step.spanContext().traceId).toBe(turn.spanContext().traceId); }); @@ -793,7 +949,7 @@ describe("createAgentOtelInstrumentation", () => { expect(rolled.spanContext().traceId).not.toBe(parentWindow.spanContext().traceId); }); - it("marks a failed SDK tool call without failing its turn", async () => { + it("marks a failed action without failing its turn", async () => { const runtime = createRuntime(); await emitAttempt({ hooks: runtime.hooks, @@ -806,7 +962,7 @@ describe("createAgentOtelInstrumentation", () => { await runtime.provider.forceFlush(); const spans = runtime.exporter.getFinishedSpans(); - expect(byName(spans, "ai.toolCall")[0]!.status.code).toBe(SpanStatusCode.ERROR); + expect(byName(spans, "agent.action")[0]!.status.code).toBe(SpanStatusCode.ERROR); expect(byName(spans, "agent.turn")[0]!.status.code).toBe(SpanStatusCode.UNSET); }); }); diff --git a/packages/eve/src/tracing/agent-otel-provider.ts b/packages/eve/src/tracing/agent-otel-provider.ts index 4eb4655ee6..0839dd6013 100644 --- a/packages/eve/src/tracing/agent-otel-provider.ts +++ b/packages/eve/src/tracing/agent-otel-provider.ts @@ -22,6 +22,7 @@ import { toolResultsContentAttribute, } from "#tracing/agent-otel-content.js"; import type { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js"; +import { createAgentActionInstrumentation } from "#tracing/agent-action-instrumentation.js"; import type { InstrumentationStepAttemptMetadataEvent, InstrumentationAttemptScope, @@ -92,6 +93,20 @@ export function createAgentOtelInstrumentation( const steps = new WeakMap(); const modelSpans = new WeakMap>(); const toolSpans = new WeakMap>(); + const actions = createAgentActionInstrumentation({ + frameworkVersion: input.frameworkVersion, + idGenerator: input.idGenerator, + recordInputs, + recordOutputs, + resolveParent: (event) => { + const step = steps.get(event.scope)?.step; + return step === undefined + ? undefined + : { context: step.context, spanContext: step.span.spanContext() }; + }, + stateStore: input.stateStore, + tracer: input.tracer, + }); const onSessionStarted = async (event: InstrumentationSessionStartedEvent): Promise => { await ensureSessionContext(event); @@ -197,6 +212,9 @@ export function createAgentOtelInstrumentation( }; const onTurnTerminal = async (event: InstrumentationTurnTerminalEvent): Promise => { + if (event.type === "turn.cancelled" || event.type === "turn.failed") { + await actions.deleteForTurn(event.sessionId, event.turnId); + } const turn = await input.stateStore.getTurn(event.sessionId, event.turnId); if (turn === undefined) return; await input.stateStore.setTurn(event.sessionId, event.turnId, { @@ -256,6 +274,7 @@ export function createAgentOtelInstrumentation( // turn that still needs its metadata — so only release session-scoped // state on terminal transitions. if (event.type === "session.completed" || event.type === "session.failed") { + await actions.deleteForSession(event.sessionId); await input.stateStore.deleteSession(event.sessionId); } }; @@ -340,9 +359,12 @@ export function createAgentOtelInstrumentation( state.span.end(); }; - const onToolCallStarted = (event: InstrumentationToolCallStartedEvent): void => { + const onToolCallStarted = async (event: InstrumentationToolCallStartedEvent): Promise => { const attempt = steps.get(event.scope); - if (attempt === undefined) return; + const parentContext = + (await actions.contextFor(event.scope.sessionId, event.scope.turnId, event.callId)) ?? + attempt?.step.context; + if (parentContext === undefined) return; const span = input.tracer.startSpan( "ai.toolCall", { @@ -352,16 +374,13 @@ export function createAgentOtelInstrumentation( "gen_ai.tool.name": event.toolName, }, }, - attempt.step.context, + parentContext, ); if (recordInputs) { const args = contentAttribute(event.input, false); if (args !== undefined) span.setAttribute("gen_ai.tool.call.arguments", args); } - const state: ToolSpanState = { - context: trace.setSpan(attempt.step.context, span), - span, - }; + const state = { context: trace.setSpan(parentContext, span), span }; getExecutionContexts(event.scope).tools.set(event.idempotencyKey, state.context); getSpanStates(toolSpans, event.scope).set(event.idempotencyKey, state); }; @@ -472,6 +491,7 @@ export function createAgentOtelInstrumentation( return { hook: { events: { + ...actions.events, "step.attempt.completed": onStepTerminal, "step.attempt.failed": onStepTerminal, "step.attempt.metadata": onStepMetadata, diff --git a/packages/eve/src/tracing/agent-trace-context-store.ts b/packages/eve/src/tracing/agent-trace-context-store.ts index 3c6111044a..c4985f2b8c 100644 --- a/packages/eve/src/tracing/agent-trace-context-store.ts +++ b/packages/eve/src/tracing/agent-trace-context-store.ts @@ -4,12 +4,14 @@ import { contextStorage, loadContext } from "#context/container.js"; import { ContextKey } from "#context/key.js"; import type { InstrumentationParentLineage } from "#harness/instrumentation-lifecycle.js"; import type { + AgentActionTraceState, AgentSessionTraceState, AgentTraceStateStore, AgentTurnTraceState, } from "#tracing/agent-trace-state.js"; interface AgentTraceContextState { + readonly actions: Readonly>; readonly sessions: Readonly>; readonly turns: Readonly>; } @@ -48,6 +50,26 @@ export function readSessionTraceContext( /** Durable trace state backed by eve's serialized Workflow context. */ export class ContextAgentTraceStateStore implements AgentTraceStateStore { + deleteAction(idempotencyKey: string): void { + updateState((state) => { + const actions = { ...state.actions }; + delete actions[idempotencyKey]; + return { ...state, actions }; + }); + } + + deleteActions(sessionId: string, turnId?: string): void { + updateState((state) => { + const actions = { ...state.actions }; + for (const [key, action] of Object.entries(actions)) { + if (action.sessionId === sessionId && (turnId === undefined || action.turnId === turnId)) { + delete actions[key]; + } + } + return { ...state, actions }; + }); + } + deleteSession(sessionId: string): void { updateState((state) => { const sessions = { ...state.sessions }; @@ -64,6 +86,16 @@ export class ContextAgentTraceStateStore implements AgentTraceStateStore { }); } + findAction(sessionId: string, callId: string): AgentActionTraceState | undefined { + return Object.values(contextStorage.getStore()?.get(AgentTraceContextKey)?.actions ?? {}).find( + (state) => state.sessionId === sessionId && state.callId === callId, + ); + } + + getAction(idempotencyKey: string): AgentActionTraceState | undefined { + return contextStorage.getStore()?.get(AgentTraceContextKey)?.actions[idempotencyKey]; + } + getSession(sessionId: string): AgentSessionTraceState | undefined { return contextStorage.getStore()?.get(AgentTraceContextKey)?.sessions[sessionId]; } @@ -72,6 +104,13 @@ export class ContextAgentTraceStateStore implements AgentTraceStateStore { return contextStorage.getStore()?.get(AgentTraceContextKey)?.turns[turnKey(sessionId, turnId)]; } + setAction(idempotencyKey: string, value: AgentActionTraceState): void { + updateState((state) => ({ + ...state, + actions: { ...state.actions, [idempotencyKey]: value }, + })); + } + setSession(sessionId: string, value: AgentSessionTraceState): void { updateState((state) => ({ ...state, @@ -88,7 +127,9 @@ export class ContextAgentTraceStateStore implements AgentTraceStateStore { } function updateState(update: (state: AgentTraceContextState) => AgentTraceContextState): void { - loadContext().set(AgentTraceContextKey, (state) => update(state ?? { sessions: {}, turns: {} })); + loadContext().set(AgentTraceContextKey, (state) => + update(state ?? { actions: {}, sessions: {}, turns: {} }), + ); } function turnKey(sessionId: string, turnId: string): string { @@ -97,6 +138,7 @@ function turnKey(sessionId: string, turnId: string): string { function serializeState(state: AgentTraceContextState): unknown { return { + actions: state.actions, sessions: Object.fromEntries( Object.entries(state.sessions).map(([id, value]) => [ id, @@ -122,7 +164,8 @@ function serializeState(state: AgentTraceContextState): unknown { } function deserializeState(data: unknown): AgentTraceContextState { - if (!isRecord(data)) return { sessions: {}, turns: {} }; + if (!isRecord(data)) return { actions: {}, sessions: {}, turns: {} }; + const actions = deserializeRecord(data.actions, deserializeAction); const sessions = deserializeRecord(data.sessions, (value) => { if (!isRecord(value) || !isSpanContext(value.context)) return undefined; return { @@ -149,7 +192,49 @@ function deserializeState(data: unknown): AgentTraceContextState { terminal: deserializeTerminal(value.terminal), } satisfies AgentTurnTraceState; }); - return { sessions, turns }; + return { actions, sessions, turns }; +} + +function deserializeAction(value: unknown): AgentActionTraceState | undefined { + if ( + !isRecord(value) || + typeof value.attemptIndex !== "number" || + typeof value.callId !== "string" || + !isActionKind(value.kind) || + typeof value.name !== "string" || + !isSpanContext(value.parent) || + typeof value.rootSessionId !== "string" || + typeof value.sessionId !== "string" || + typeof value.spanId !== "string" || + typeof value.startTimeMs !== "number" || + typeof value.stepIndex !== "number" || + typeof value.turnId !== "string" + ) { + return undefined; + } + return { + attemptIndex: value.attemptIndex, + callId: value.callId, + inputAttribute: typeof value.inputAttribute === "string" ? value.inputAttribute : undefined, + kind: value.kind, + name: value.name, + parent: value.parent, + rootSessionId: value.rootSessionId, + sessionId: value.sessionId, + spanId: value.spanId, + startTimeMs: value.startTimeMs, + stepIndex: value.stepIndex, + turnId: value.turnId, + }; +} + +function isActionKind(value: unknown): value is AgentActionTraceState["kind"] { + return ( + value === "load-skill" || + value === "remote-agent-call" || + value === "subagent-call" || + value === "tool-call" + ); } function deserializeRecord( diff --git a/packages/eve/src/tracing/agent-trace-state.ts b/packages/eve/src/tracing/agent-trace-state.ts index 93e9f20b6a..49c7c5cbf9 100644 --- a/packages/eve/src/tracing/agent-trace-state.ts +++ b/packages/eve/src/tracing/agent-trace-state.ts @@ -1,7 +1,9 @@ import type { SpanContext } from "#compiled/@opentelemetry/api/index.js"; import type { + InstrumentationActionKind, InstrumentationParentLineage, + InstrumentationTraceContext, InstrumentationTurnFailedEvent, InstrumentationTurnSettledEvent, } from "#harness/instrumentation-lifecycle.js"; @@ -30,10 +32,34 @@ export interface AgentTurnTraceState { | { readonly type: InstrumentationTurnSettledEvent["type"] }; } +export interface AgentActionTraceState { + readonly attemptIndex: number; + readonly callId: string; + readonly inputAttribute?: string; + readonly kind: InstrumentationActionKind; + readonly name: string; + readonly parent: InstrumentationTraceContext; + readonly rootSessionId: string; + readonly sessionId: string; + readonly spanId: string; + readonly startTimeMs: number; + readonly stepIndex: number; + readonly turnId: string; +} + /** Provider-owned serializable storage for durable agent trace state. */ export interface AgentTraceStateStore { + deleteAction(idempotencyKey: string): void | PromiseLike; + deleteActions(sessionId: string, turnId?: string): void | PromiseLike; deleteSession(sessionId: string): void | PromiseLike; deleteTurn(sessionId: string, turnId: string): void | PromiseLike; + findAction( + sessionId: string, + callId: string, + ): AgentActionTraceState | undefined | PromiseLike; + getAction( + idempotencyKey: string, + ): AgentActionTraceState | undefined | PromiseLike; getSession( sessionId: string, ): AgentSessionTraceState | undefined | PromiseLike; @@ -41,15 +67,29 @@ export interface AgentTraceStateStore { sessionId: string, turnId: string, ): AgentTurnTraceState | undefined | PromiseLike; + setAction(idempotencyKey: string, state: AgentActionTraceState): void | PromiseLike; setSession(sessionId: string, state: AgentSessionTraceState): void | PromiseLike; setTurn(sessionId: string, turnId: string, state: AgentTurnTraceState): void | PromiseLike; } /** In-memory trace state used by tests and non-durable runtimes. */ export class InMemoryAgentTraceStateStore implements AgentTraceStateStore { + readonly #actions = new Map(); readonly #sessions = new Map(); readonly #turns = new Map(); + deleteAction(idempotencyKey: string): void { + this.#actions.delete(idempotencyKey); + } + + deleteActions(sessionId: string, turnId?: string): void { + for (const [key, state] of this.#actions) { + if (state.sessionId === sessionId && (turnId === undefined || state.turnId === turnId)) { + this.#actions.delete(key); + } + } + } + deleteSession(sessionId: string): void { this.#sessions.delete(sessionId); } @@ -58,6 +98,16 @@ export class InMemoryAgentTraceStateStore implements AgentTraceStateStore { this.#turns.delete(turnKey(sessionId, turnId)); } + findAction(sessionId: string, callId: string): AgentActionTraceState | undefined { + return [...this.#actions.values()].find( + (state) => state.sessionId === sessionId && state.callId === callId, + ); + } + + getAction(idempotencyKey: string): AgentActionTraceState | undefined { + return this.#actions.get(idempotencyKey); + } + getSession(sessionId: string): AgentSessionTraceState | undefined { return this.#sessions.get(sessionId); } @@ -66,6 +116,10 @@ export class InMemoryAgentTraceStateStore implements AgentTraceStateStore { return this.#turns.get(turnKey(sessionId, turnId)); } + setAction(idempotencyKey: string, state: AgentActionTraceState): void { + this.#actions.set(idempotencyKey, state); + } + setSession(sessionId: string, state: AgentSessionTraceState): void { this.#sessions.set(sessionId, state); } 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 97c9180560..99185b6bff 100644 --- a/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts +++ b/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts @@ -11,6 +11,7 @@ import { createAiSdkHookBridge } from "#harness/ai-sdk-hook-bridge.js"; import { listLocalTraces } from "#tracing/local-trace-reader.js"; import type { InstrumentationAttemptScope } from "#harness/instrumentation-lifecycle.js"; import { + actionIdempotencyKey, attemptIdempotencyKey, sessionIdempotencyKey, turnIdempotencyKey, @@ -92,6 +93,16 @@ describe("local instrumentation runtime", () => { usage: { inputTokens: 1, outputTokens: 1 }, }, ]); + const actionKey = actionIdempotencyKey("session-1", "turn-1", "tool-1"); + await runtime.hooks.publish({ + callId: "tool-1", + idempotencyKey: actionKey, + input: {}, + kind: "tool-call", + name: "weather", + scope, + type: "action.started", + }); await Reflect.apply(bridge.onToolExecutionStart!, bridge, [ { callId: "call-1", @@ -116,6 +127,12 @@ describe("local instrumentation runtime", () => { toolOutput: { output: { temperature: 72 }, type: "tool-result" }, }, ]); + await runtime.hooks.publish({ + idempotencyKey: actionKey, + output: { output: { temperature: 72 }, type: "result" }, + scope, + type: "action.completed", + }); await runtime.hooks.publish({ idempotencyKey: attemptIdempotencyKey(scope), scope, @@ -157,6 +174,7 @@ describe("local instrumentation runtime", () => { "agent.step", "ai.streamText", "ai.streamText.doStream", + "agent.action", "ai.toolCall", "user.model-work", "user.tool-work", @@ -170,7 +188,8 @@ describe("local instrumentation runtime", () => { expect(span(spans, "user.model-work").parentSpanId).toBe( span(spans, "ai.streamText.doStream").spanId, ); - expect(span(spans, "ai.toolCall").parentSpanId).toBe(span(spans, "agent.step").spanId); + expect(span(spans, "agent.action").parentSpanId).toBe(span(spans, "agent.step").spanId); + expect(span(spans, "ai.toolCall").parentSpanId).toBe(span(spans, "agent.action").spanId); expect(span(spans, "user.tool-work").parentSpanId).toBe(span(spans, "ai.toolCall").spanId); const listed = await listLocalTraces(appRoot); expect(listed).toHaveLength(1);