From fcddb1b01149ee029701eaf9c9c7abcee1224554 Mon Sep 17 00:00:00 2001 From: Chad Hietala Date: Sat, 8 Aug 2026 16:56:05 -0400 Subject: [PATCH] feat(eve): key instrumentation events by identity Signed-off-by: Chad Hietala --- .changeset/instrumentation-event-identity.md | 5 + .../src/harness/ai-sdk-hook-bridge.test.ts | 57 +++++++---- .../eve/src/harness/ai-sdk-hook-bridge.ts | 99 ++++++++++--------- .../harness/instrumentation-lifecycle.test.ts | 40 ++++++++ .../src/harness/instrumentation-lifecycle.ts | 98 ++++++++++++++---- .../instrumentation-native-events.test.ts | 15 ++- .../harness/instrumentation-native-events.ts | 25 ++++- .../harness/instrumentation-providers.test.ts | 2 + packages/eve/src/harness/tool-loop.ts | 3 + .../src/public/instrumentation/provider.ts | 6 ++ .../src/tracing/agent-otel-provider.test.ts | 30 +++++- .../eve/src/tracing/agent-otel-provider.ts | 27 ++--- .../tracing/agent-trace-context-store.test.ts | 4 +- .../src/tracing/agent-trace-context-store.ts | 9 +- packages/eve/src/tracing/agent-trace-state.ts | 10 +- ...l-instrumentation-runtime.scenario.test.ts | 15 ++- 16 files changed, 330 insertions(+), 115 deletions(-) create mode 100644 .changeset/instrumentation-event-identity.md create mode 100644 packages/eve/src/harness/instrumentation-lifecycle.test.ts diff --git a/.changeset/instrumentation-event-identity.md b/.changeset/instrumentation-event-identity.md new file mode 100644 index 0000000000..4cb8834dc8 --- /dev/null +++ b/.changeset/instrumentation-event-identity.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Give instrumentation lifecycle events replay-stable idempotency keys so providers can update one record across retries and worker replays. diff --git a/packages/eve/src/harness/ai-sdk-hook-bridge.test.ts b/packages/eve/src/harness/ai-sdk-hook-bridge.test.ts index 7b5c4ae998..a04e48dca7 100644 --- a/packages/eve/src/harness/ai-sdk-hook-bridge.test.ts +++ b/packages/eve/src/harness/ai-sdk-hook-bridge.test.ts @@ -29,11 +29,13 @@ describe("createAiSdkHookBridge", () => { return { events: { "model.call.started"(event) { - calls.push(`${name}:started:${event.id}`); - states.set(event.id, `${name}-state`); + calls.push(`${name}:started:${event.idempotencyKey}`); + states.set(event.idempotencyKey, `${name}-state`); }, "model.call.completed"(event) { - calls.push(`${name}:completed:${event.id}:${String(states.get(event.id))}`); + calls.push( + `${name}:completed:${event.idempotencyKey}:${String(states.get(event.idempotencyKey))}`, + ); }, }, }; @@ -55,7 +57,7 @@ describe("createAiSdkHookBridge", () => { }, ]); - const id = `${scope.attemptId}:model:call-1:0`; + const id = `model:${scope.attemptId}:0`; expect(calls).toEqual([ `a:started:${id}`, `b:started:${id}`, @@ -91,10 +93,10 @@ describe("createAiSdkHookBridge", () => { it("passes the identity captured at model-call start to the context runner", async () => { const ids: string[] = []; const hooks = createInstrumentationHooks([ - { events: { "model.call.started": (event) => void ids.push(event.id) } }, + { events: { "model.call.started": (event) => void ids.push(event.idempotencyKey) } }, ]); const bridge = createAiSdkHookBridge(scope, hooks, (operation, execute) => { - ids.push(operation.id); + ids.push(operation.idempotencyKey); return execute(); }); Reflect.apply(bridge.onStart!, bridge, [ @@ -108,7 +110,7 @@ describe("createAiSdkHookBridge", () => { await bridge.executeLanguageModelCall!({ callId: "call-1", execute: async () => "result" }); - const expected = `${scope.attemptId}:model:call-1:0`; + const expected = `model:${scope.attemptId}:0`; expect(ids).toEqual([expected, expected]); }); @@ -126,6 +128,23 @@ describe("createAiSdkHookBridge", () => { expect(adapterCalls).toBe(0); }); + it("derives replay-stable model identity without the AI SDK call ID", async () => { + const keys: string[] = []; + const hooks = createInstrumentationHooks([ + { events: { "model.call.started": (event) => void keys.push(event.idempotencyKey) } }, + ]); + + for (const callId of ["sdk-random-1", "sdk-random-2"]) { + const bridge = createAiSdkHookBridge(scope, hooks); + await Reflect.apply(bridge.onStepStart!, bridge, [{ callId, stepNumber: 2 }]); + await Reflect.apply(bridge.onLanguageModelCallStart!, bridge, [ + { callId, messages: [], modelId: "model", provider: "test", tools: undefined }, + ]); + } + + expect(keys).toEqual([`model:${scope.attemptId}:2`, `model:${scope.attemptId}:2`]); + }); + it("publishes step provider metadata as step.metadata, skipping steps without any", async () => { const events: InstrumentationStepAttemptMetadataEvent[] = []; const hooks = createInstrumentationHooks([ @@ -146,6 +165,7 @@ describe("createAiSdkHookBridge", () => { expect(events).toEqual([ { + idempotencyKey: `step:${scope.attemptId}`, providerMetadata: { gateway: { cost: "0.000082" } }, scope, type: "step.attempt.metadata", @@ -225,6 +245,7 @@ describe("createAiSdkHookBridge", () => { await Reflect.apply(bridge.onStepStart!, bridge, [{ callId: "call-1", stepNumber: 0 }]); const expected = { + idempotencyKey: `step:${scope.attemptId}`, operation: { modelId: "model", operationId: "ai.streamText", provider: "test" }, scope, type: "step.attempt.started", @@ -286,7 +307,7 @@ describe("createAiSdkHookBridge", () => { ]); expect(before).toHaveBeenCalledExactlyOnceWith({ - id: `${scope.attemptId}:model:call-1:0`, + idempotencyKey: `model:${scope.attemptId}:0`, input: { instructions: "be brief", messages: [{ content: "hi", role: "user" }] }, model: { modelId: "model", provider: "test" }, scope, @@ -303,7 +324,7 @@ describe("createAiSdkHookBridge", () => { { error: "boom", input: { a: 2 }, toolName: "search", type: "tool-error" }, ], finishReason: "tool-calls", - id: `${scope.attemptId}:model:call-1:0`, + idempotencyKey: `model:${scope.attemptId}:0`, scope, type: "model.call.completed", usage: { @@ -347,7 +368,7 @@ describe("createAiSdkHookBridge", () => { expect(before).toHaveBeenCalledExactlyOnceWith({ callId: "tool-1", - id: `${scope.attemptId}:tool:tool-1:0`, + idempotencyKey: `tool:${scope.attemptId}:tool-1:0`, input: { q: "eve" }, kind: "tool-call", scope, @@ -355,7 +376,7 @@ describe("createAiSdkHookBridge", () => { type: "tool.call.started", }); expect(after).toHaveBeenCalledExactlyOnceWith({ - id: `${scope.attemptId}:tool:tool-1:0`, + idempotencyKey: `tool:${scope.attemptId}:tool-1:0`, output: expected, scope, type: "tool.call.completed", @@ -389,9 +410,9 @@ describe("createAiSdkHookBridge", () => { return { events: { "model.call.completed": (event) => { - observed.set(name, own.get(event.id)); + observed.set(name, own.get(event.idempotencyKey)); }, - "model.call.started": (event) => void own.set(event.id, `${name}-state`), + "model.call.started": (event) => void own.set(event.idempotencyKey, `${name}-state`), }, }; }; @@ -450,14 +471,14 @@ describe("createAiSdkHookBridge", () => { events: { async "tool.call.started"(event) { started.set( - event.id, + event.idempotencyKey, await new Promise((resolve) => { - resolvers.set(event.id, () => resolve(`state:${event.id}`)); + resolvers.set(event.idempotencyKey, () => resolve(`state:${event.idempotencyKey}`)); }), ); }, "tool.call.completed"(event) { - terminalStates.set(event.id, started.get(event.id)); + terminalStates.set(event.idempotencyKey, started.get(event.idempotencyKey)); }, }, }, @@ -474,8 +495,8 @@ describe("createAiSdkHookBridge", () => { const second = start("tool-2"); await vi.waitFor(() => expect(resolvers.size).toBe(2)); - const firstId = `${scope.attemptId}:tool:tool-1:0`; - const secondId = `${scope.attemptId}:tool:tool-2:0`; + const firstId = `tool:${scope.attemptId}:tool-1:0`; + const secondId = `tool:${scope.attemptId}:tool-2:0`; resolvers.get(secondId)!(); resolvers.get(firstId)!(); await Promise.all([first, second]); diff --git a/packages/eve/src/harness/ai-sdk-hook-bridge.ts b/packages/eve/src/harness/ai-sdk-hook-bridge.ts index a9e777b25a..9ebee70814 100644 --- a/packages/eve/src/harness/ai-sdk-hook-bridge.ts +++ b/packages/eve/src/harness/ai-sdk-hook-bridge.ts @@ -15,6 +15,11 @@ import type { InstrumentationToolOutput, InstrumentationUsage, } from "#harness/instrumentation-lifecycle.js"; +import { + attemptIdempotencyKey, + modelCallIdempotencyKey, + toolCallIdempotencyKey, +} from "#harness/instrumentation-lifecycle.js"; type TelemetryEvent = Parameters>[0]; @@ -25,10 +30,10 @@ type TelemetryEvent = Parameters InstrumentationActionKind; interface AttemptState { - readonly modelIds: Map; + readonly modelKeys: Map; readonly resolveActionKind: ActionKindResolver; readonly scope: InstrumentationAttemptScope; - readonly toolIds: Map; + readonly toolKeys: Map; operation?: InstrumentationOperationRef; // Only the number is kept: it disambiguates call identities within an attempt. stepNumber?: number; @@ -42,10 +47,10 @@ export function createAiSdkHookBridge( resolveActionKind: ActionKindResolver = defaultResolveActionKind, ): Telemetry { const state: AttemptState = { - modelIds: new Map(), + modelKeys: new Map(), resolveActionKind, scope, - toolIds: new Map(), + toolKeys: new Map(), }; return { @@ -62,22 +67,22 @@ export function createAiSdkHookBridge( if (started !== undefined) await hooks.publish(started); }, async onLanguageModelCallStart(event) { - const id = createModelCallIdentity(state, event.callId); - state.modelIds.set(event.callId, id); - const started = toModelCallStarted(state, id, event); + const key = modelCallIdempotencyKey(state.scope, state.stepNumber ?? 0); + state.modelKeys.set(event.callId, key); + const started = toModelCallStarted(state, key, event); await hooks.publish(started); }, executeLanguageModelCall({ callId, execute }) { - const id = state.modelIds.get(callId); - return id === undefined + const key = state.modelKeys.get(callId); + return key === undefined ? execute() - : runInContext({ id, scope, type: "model.call" }, execute); + : runInContext({ idempotencyKey: key, scope, type: "model.call" }, execute); }, async onLanguageModelCallEnd(event) { - const id = state.modelIds.get(event.callId); - if (id === undefined) return; - state.modelIds.delete(event.callId); - const completed = toModelCallCompleted(state, id, event); + const key = state.modelKeys.get(event.callId); + if (key === undefined) return; + state.modelKeys.delete(event.callId); + const completed = toModelCallCompleted(state, key, event); await hooks.publish(completed); }, async onStepEnd(event) { @@ -87,6 +92,7 @@ export function createAiSdkHookBridge( if (event.providerMetadata === undefined) return; await hooks.publish( Object.freeze({ + idempotencyKey: attemptIdempotencyKey(state.scope), providerMetadata: event.providerMetadata, scope: state.scope, type: "step.attempt.metadata", @@ -94,21 +100,27 @@ export function createAiSdkHookBridge( ); }, async onToolExecutionStart(event) { - const id = createToolCallIdentity(state, event.toolCall.toolCallId); - state.toolIds.set(event.toolCall.toolCallId, id); - const started = toToolCallStarted(state, id, event); + const key = toolCallIdempotencyKey( + state.scope, + event.toolCall.toolCallId, + state.stepNumber ?? 0, + ); + state.toolKeys.set(event.toolCall.toolCallId, key); + const started = toToolCallStarted(state, key, event); await hooks.publish(started); }, executeTool({ toolCallId, execute }) { - const id = state.toolIds.get(toolCallId); - return id === undefined ? execute() : runInContext({ id, scope, type: "tool.call" }, execute); + const key = state.toolKeys.get(toolCallId); + return key === undefined + ? execute() + : runInContext({ idempotencyKey: key, scope, type: "tool.call" }, execute); }, async onToolExecutionEnd(event) { const toolCallId = event.toolCall.toolCallId; - const id = state.toolIds.get(toolCallId); - if (id === undefined) return; - state.toolIds.delete(toolCallId); - const completed = toToolCallCompleted(state, id, event); + const key = state.toolKeys.get(toolCallId); + if (key === undefined) return; + state.toolKeys.delete(toolCallId); + const completed = toToolCallCompleted(state, key, event); await hooks.publish(completed); }, async onAbort(event) { @@ -121,14 +133,18 @@ export function createAiSdkHookBridge( async function failOpenOperations(error: unknown): Promise { const pending: Promise[] = []; - for (const id of state.modelIds.values()) { - pending.push(hooks.publish(Object.freeze({ error, id, scope, type: "model.call.failed" }))); + for (const idempotencyKey of state.modelKeys.values()) { + pending.push( + hooks.publish(Object.freeze({ error, idempotencyKey, scope, type: "model.call.failed" })), + ); } - for (const id of state.toolIds.values()) { - pending.push(hooks.publish(Object.freeze({ error, id, scope, type: "tool.call.failed" }))); + for (const idempotencyKey of state.toolKeys.values()) { + pending.push( + hooks.publish(Object.freeze({ error, idempotencyKey, scope, type: "tool.call.failed" })), + ); } - state.modelIds.clear(); - state.toolIds.clear(); + state.modelKeys.clear(); + state.toolKeys.clear(); await Promise.all(pending); } } @@ -142,23 +158,20 @@ function toStepAttemptStarted( ): InstrumentationStepAttemptStartedEvent | undefined { if (state.operation === undefined || state.stepNumber === undefined) return undefined; return Object.freeze({ + idempotencyKey: attemptIdempotencyKey(state.scope), operation: state.operation, scope: state.scope, type: "step.attempt.started", }); } -function createModelCallIdentity(state: AttemptState, callId: string): string { - return `${state.scope.attemptId}:model:${callId}:${state.stepNumber ?? 0}`; -} - function toModelCallStarted( state: AttemptState, - id: string, + idempotencyKey: string, source: TelemetryEvent<"onLanguageModelCallStart">, ): InstrumentationModelCallStartedEvent { return Object.freeze({ - id, + idempotencyKey, input: Object.freeze({ instructions: source.instructions, messages: Object.freeze([...source.messages]), @@ -171,13 +184,13 @@ function toModelCallStarted( function toModelCallCompleted( state: AttemptState, - id: string, + idempotencyKey: string, source: TelemetryEvent<"onLanguageModelCallEnd">, ): InstrumentationModelCallCompletedEvent { return Object.freeze({ content: toContentParts(source.content), finishReason: source.finishReason, - id, + idempotencyKey, scope: state.scope, type: "model.call.completed", usage: toUsage(source.usage), @@ -238,18 +251,14 @@ function toContentParts( return Object.freeze(parts); } -function createToolCallIdentity(state: AttemptState, toolCallId: string): string { - return `${state.scope.attemptId}:tool:${toolCallId}:${state.stepNumber ?? 0}`; -} - function toToolCallStarted( state: AttemptState, - id: string, + idempotencyKey: string, source: TelemetryEvent<"onToolExecutionStart">, ): InstrumentationToolCallStartedEvent { return Object.freeze({ callId: source.toolCall.toolCallId, - id, + idempotencyKey, input: source.toolCall.input, kind: state.resolveActionKind(source.toolCall.toolName), scope: state.scope, @@ -260,11 +269,11 @@ function toToolCallStarted( function toToolCallCompleted( state: AttemptState, - id: string, + idempotencyKey: string, source: TelemetryEvent<"onToolExecutionEnd">, ): InstrumentationToolCallCompletedEvent { return Object.freeze({ - id, + idempotencyKey, output: toToolOutput(source.toolOutput), scope: state.scope, type: "tool.call.completed", diff --git a/packages/eve/src/harness/instrumentation-lifecycle.test.ts b/packages/eve/src/harness/instrumentation-lifecycle.test.ts new file mode 100644 index 0000000000..1771d53c42 --- /dev/null +++ b/packages/eve/src/harness/instrumentation-lifecycle.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, it } from "vitest"; + +import { + attemptIdempotencyKey, + modelCallIdempotencyKey, + sessionIdempotencyKey, + toolCallIdempotencyKey, + turnIdempotencyKey, + type InstrumentationAttemptScope, +} from "#harness/instrumentation-lifecycle.js"; + +const scope: InstrumentationAttemptScope = { + attemptId: "session-1:turn-1:0:0", + attemptIndex: 0, + sessionId: "session-1", + stepIndex: 0, + turnId: "turn-1", +}; + +describe("instrumentation idempotency keys", () => { + it("separates classes that share an identifier", () => { + expect(sessionIdempotencyKey("shared")).not.toBe(turnIdempotencyKey("shared", "shared")); + }); + + it("separates identical turn IDs in different sessions", () => { + expect(turnIdempotencyKey("session-1", "turn_0")).not.toBe( + turnIdempotencyKey("session-2", "turn_0"), + ); + }); + + it("derives model identity without an AI SDK call ID", () => { + expect(modelCallIdempotencyKey(scope, 2)).toBe("model:session-1:turn-1:0:0:2"); + }); + + it("separates model attempts and SDK tool calls", () => { + const retry = { ...scope, attemptId: "session-1:turn-1:0:1", attemptIndex: 1 }; + expect(attemptIdempotencyKey(scope)).not.toBe(attemptIdempotencyKey(retry)); + expect(modelCallIdempotencyKey(scope, 0)).not.toBe(toolCallIdempotencyKey(scope, "call-1", 0)); + }); +}); diff --git a/packages/eve/src/harness/instrumentation-lifecycle.ts b/packages/eve/src/harness/instrumentation-lifecycle.ts index 78c501ac7b..d974a04a46 100644 --- a/packages/eve/src/harness/instrumentation-lifecycle.ts +++ b/packages/eve/src/harness/instrumentation-lifecycle.ts @@ -81,8 +81,38 @@ export type InstrumentationToolOutput = | { readonly type: "result"; readonly output: unknown } | { readonly type: "error"; readonly error: unknown }; +/** Replay-stable row identity for every lifecycle operation. */ +export function sessionIdempotencyKey(sessionId: string): string { + return `session:${sessionId}`; +} + +export function turnIdempotencyKey(sessionId: string, turnId: string): string { + return `turn:${sessionId}:${turnId}`; +} + +export function attemptIdempotencyKey(scope: InstrumentationAttemptScope): string { + return `step:${scope.attemptId}`; +} + +/** One model call occurs per AI SDK step within an eve attempt. */ +export function modelCallIdempotencyKey( + scope: InstrumentationAttemptScope, + stepNumber: number, +): string { + return `model:${scope.attemptId}:${String(stepNumber)}`; +} + +export function toolCallIdempotencyKey( + scope: InstrumentationAttemptScope, + callId: string, + stepNumber: number, +): string { + return `tool:${scope.attemptId}:${callId}:${String(stepNumber)}`; +} + export interface InstrumentationStepAttemptStartedEvent { readonly type: "step.attempt.started"; + readonly idempotencyKey: string; readonly operation: InstrumentationOperationRef; readonly scope: InstrumentationAttemptScope; } @@ -91,6 +121,7 @@ export interface InstrumentationSessionStartedEvent { readonly type: "session.started"; readonly agentName?: string; readonly channelKind?: string; + readonly idempotencyKey: string; readonly parentTraceContext?: InstrumentationTraceContext; readonly rootSessionId: string; readonly sessionId: string; @@ -122,6 +153,7 @@ export interface InstrumentationParentLineage { */ export interface InstrumentationSessionSettledEvent { readonly type: "session.completed" | "session.waiting"; + readonly idempotencyKey: string; readonly sessionId: string; readonly turnId?: string; } @@ -129,6 +161,7 @@ export interface InstrumentationSessionSettledEvent { export interface InstrumentationSessionFailedEvent { readonly type: "session.failed"; readonly error: unknown; + readonly idempotencyKey: string; readonly sessionId: string; readonly turnId?: string; } @@ -139,6 +172,7 @@ export type InstrumentationSessionTransitionEvent = export interface InstrumentationTurnStartedEvent { readonly type: "turn.started"; + readonly idempotencyKey: string; readonly parentLineage?: InstrumentationParentLineage; readonly parentTraceContext?: InstrumentationTraceContext; readonly rootSessionId: string; @@ -147,19 +181,42 @@ export interface InstrumentationTurnStartedEvent { readonly turnId: string; } -export interface InstrumentationTurnTerminalEvent { - readonly type: "turn.cancelled" | "turn.completed" | "turn.failed"; - readonly error?: unknown; +export interface InstrumentationTurnSettledEvent { + readonly type: "turn.cancelled" | "turn.completed"; + readonly idempotencyKey: string; readonly sessionId: string; readonly turnId: string; } -export interface InstrumentationStepAttemptTerminalEvent { - readonly type: "step.attempt.completed" | "step.attempt.failed"; - readonly error?: unknown; +export interface InstrumentationTurnFailedEvent { + readonly type: "turn.failed"; + readonly error: unknown; + readonly idempotencyKey: string; + readonly sessionId: string; + readonly turnId: string; +} + +export type InstrumentationTurnTerminalEvent = + | InstrumentationTurnSettledEvent + | InstrumentationTurnFailedEvent; + +export interface InstrumentationStepAttemptCompletedEvent { + readonly type: "step.attempt.completed"; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; } +export interface InstrumentationStepAttemptFailedEvent { + readonly type: "step.attempt.failed"; + readonly error: unknown; + readonly idempotencyKey: string; + readonly scope: InstrumentationAttemptScope; +} + +export type InstrumentationStepAttemptTerminalEvent = + | InstrumentationStepAttemptCompletedEvent + | InstrumentationStepAttemptFailedEvent; + /** * Provider metadata for one completed attempt, as reported by the AI SDK * (`StepResult.providerMetadata`). Carries Vercel AI Gateway cost data when @@ -167,13 +224,14 @@ export interface InstrumentationStepAttemptTerminalEvent { */ export interface InstrumentationStepAttemptMetadataEvent { readonly type: "step.attempt.metadata"; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; readonly providerMetadata: Readonly>; } export interface InstrumentationModelCallStartedEvent { readonly type: "model.call.started"; - readonly id: string; + readonly idempotencyKey: string; readonly input: InstrumentationModelInput; readonly model: InstrumentationModelRef; readonly scope: InstrumentationAttemptScope; @@ -183,7 +241,7 @@ export interface InstrumentationModelCallCompletedEvent { readonly type: "model.call.completed"; readonly content: readonly InstrumentationContentPart[]; readonly finishReason: string; - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; readonly usage: InstrumentationUsage; } @@ -191,7 +249,7 @@ export interface InstrumentationModelCallCompletedEvent { export interface InstrumentationModelCallFailedEvent { readonly type: "model.call.failed"; readonly error: unknown; - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; } @@ -202,7 +260,7 @@ export type InstrumentationModelCallTerminalEvent = export interface InstrumentationToolCallStartedEvent { readonly type: "tool.call.started"; readonly callId: string; - readonly id: string; + readonly idempotencyKey: string; readonly input: unknown; readonly kind: InstrumentationActionKind; readonly scope: InstrumentationAttemptScope; @@ -211,7 +269,7 @@ export interface InstrumentationToolCallStartedEvent { export interface InstrumentationToolCallCompletedEvent { readonly type: "tool.call.completed"; - readonly id: string; + readonly idempotencyKey: string; readonly output: InstrumentationToolOutput; readonly scope: InstrumentationAttemptScope; } @@ -219,7 +277,7 @@ export interface InstrumentationToolCallCompletedEvent { export interface InstrumentationToolCallFailedEvent { readonly type: "tool.call.failed"; readonly error: unknown; - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; } @@ -238,8 +296,8 @@ export type InstrumentationEventHandler = (event: TEvent) => void | Prom export interface InstrumentationProviderDefinition { readonly events?: { readonly "step.attempt.started"?: InstrumentationEventHandler; - readonly "step.attempt.completed"?: InstrumentationEventHandler; - readonly "step.attempt.failed"?: InstrumentationEventHandler; + readonly "step.attempt.completed"?: InstrumentationEventHandler; + readonly "step.attempt.failed"?: InstrumentationEventHandler; readonly "step.attempt.metadata"?: InstrumentationEventHandler; readonly "model.call.started"?: InstrumentationEventHandler; readonly "model.call.completed"?: InstrumentationEventHandler; @@ -251,9 +309,9 @@ export interface InstrumentationProviderDefinition { readonly "tool.call.started"?: InstrumentationEventHandler; readonly "tool.call.completed"?: InstrumentationEventHandler; readonly "tool.call.failed"?: InstrumentationEventHandler; - readonly "turn.cancelled"?: InstrumentationEventHandler; - readonly "turn.completed"?: InstrumentationEventHandler; - readonly "turn.failed"?: InstrumentationEventHandler; + readonly "turn.cancelled"?: InstrumentationEventHandler; + readonly "turn.completed"?: InstrumentationEventHandler; + readonly "turn.failed"?: InstrumentationEventHandler; readonly "turn.started"?: InstrumentationEventHandler; }; readonly flush?: () => void | PromiseLike; @@ -261,7 +319,7 @@ export interface InstrumentationProviderDefinition { readonly shutdown?: () => void | PromiseLike; } -/** Events that carry an operation `id`, pairing a start with its terminal. */ +/** Events that pair a start with its terminal under one `idempotencyKey`. */ export type InstrumentationCorrelatedEvent = | InstrumentationModelCallStartedEvent | InstrumentationModelCallTerminalEvent @@ -288,12 +346,12 @@ export type InstrumentationContextRunner = ( /** Stable identity supplied only to a trusted framework context runner. */ export type InstrumentationExecutionOperation = | { - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; readonly type: "model.call"; } | { - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; readonly type: "tool.call"; }; diff --git a/packages/eve/src/harness/instrumentation-native-events.test.ts b/packages/eve/src/harness/instrumentation-native-events.test.ts index a67b8f24f1..7f87122a75 100644 --- a/packages/eve/src/harness/instrumentation-native-events.test.ts +++ b/packages/eve/src/harness/instrumentation-native-events.test.ts @@ -84,7 +84,14 @@ describe("createInstrumentationHandleEvent", () => { await handleEvent(createSessionWaitingEvent()); - expect(events).toEqual([{ sessionId: "session-1", turnId: "turn-1", type: "session.waiting" }]); + expect(events).toEqual([ + { + idempotencyKey: "session:session-1", + sessionId: "session-1", + turnId: "turn-1", + type: "session.waiting", + }, + ]); }); it("carries the dispatch lineage onto every turn a child session starts", async () => { @@ -112,6 +119,7 @@ describe("createInstrumentationHandleEvent", () => { expect(events.filter((event) => event.type === "turn.started")).toEqual([ { + idempotencyKey: "turn:child-1:child-turn-1", parentLineage, parentTraceContext: undefined, rootSessionId: "session-1", @@ -121,6 +129,7 @@ describe("createInstrumentationHandleEvent", () => { type: "turn.started", }, { + idempotencyKey: "turn:child-1:child-turn-2", parentLineage, parentTraceContext: undefined, rootSessionId: "session-1", @@ -218,7 +227,7 @@ describe("createInstrumentationHandleEvent", () => { expect(events).toEqual([ { callId: "delegate-1", - id: "session-1:turn-1:0:0:tool:delegate-1:0", + idempotencyKey: "tool:session-1:turn-1:0:0:delegate-1:0", input: { task: "research" }, kind: "subagent-call", scope, @@ -227,7 +236,7 @@ describe("createInstrumentationHandleEvent", () => { }, { callId: "remote-1", - id: "session-1:turn-1:0:0:tool:remote-1:0", + idempotencyKey: "tool:session-1:turn-1:0:0:remote-1:0", input: { task: "analyze" }, kind: "remote-agent-call", scope, diff --git a/packages/eve/src/harness/instrumentation-native-events.ts b/packages/eve/src/harness/instrumentation-native-events.ts index 42daaaab96..98161c3c6c 100644 --- a/packages/eve/src/harness/instrumentation-native-events.ts +++ b/packages/eve/src/harness/instrumentation-native-events.ts @@ -7,6 +7,11 @@ import type { InstrumentationToolCallStartedEvent, InstrumentationTraceContext, } from "#harness/instrumentation-lifecycle.js"; +import { + sessionIdempotencyKey, + toolCallIdempotencyKey, + turnIdempotencyKey, +} from "#harness/instrumentation-lifecycle.js"; import type { HandleEventFn, HarnessToolMap } from "#harness/types.js"; export interface InstrumentationActionSource { @@ -67,7 +72,7 @@ async function publishDelegationActions( await hooks.publish( Object.freeze({ callId: action.callId, - id: `${source.scope.attemptId}:tool:${action.callId}:0`, + idempotencyKey: toolCallIdempotencyKey(source.scope, action.callId, 0), input: action.input, kind: tool.runtimeAction.kind, scope: source.scope, @@ -87,6 +92,7 @@ function toLifecycleEvent( case "session.started": return { agentName: input.agentName, + idempotencyKey: sessionIdempotencyKey(input.sessionId), parentTraceContext: input.parentTraceContext, rootSessionId: input.rootSessionId ?? input.sessionId, sessionId: input.sessionId, @@ -94,16 +100,23 @@ function toLifecycleEvent( }; case "session.completed": case "session.waiting": - return { sessionId: input.sessionId, turnId: activeTurnId, type: event.type }; + return { + idempotencyKey: sessionIdempotencyKey(input.sessionId), + sessionId: input.sessionId, + turnId: activeTurnId, + type: event.type, + }; case "session.failed": return { error: new Error(event.data.message), + idempotencyKey: sessionIdempotencyKey(input.sessionId), sessionId: input.sessionId, turnId: activeTurnId, type: "session.failed", }; case "turn.started": return { + idempotencyKey: turnIdempotencyKey(input.sessionId, event.data.turnId), parentLineage: input.parentLineage, parentTraceContext: input.parentTraceContext, rootSessionId: input.rootSessionId ?? input.sessionId, @@ -114,10 +127,16 @@ function toLifecycleEvent( }; case "turn.completed": case "turn.cancelled": - return { sessionId: input.sessionId, turnId: event.data.turnId, type: event.type }; + return { + idempotencyKey: turnIdempotencyKey(input.sessionId, event.data.turnId), + sessionId: input.sessionId, + turnId: event.data.turnId, + type: event.type, + }; case "turn.failed": return { error: new Error(event.data.message), + idempotencyKey: turnIdempotencyKey(input.sessionId, event.data.turnId), sessionId: input.sessionId, turnId: event.data.turnId, type: "turn.failed", diff --git a/packages/eve/src/harness/instrumentation-providers.test.ts b/packages/eve/src/harness/instrumentation-providers.test.ts index 893aa11248..5e75e09f0e 100644 --- a/packages/eve/src/harness/instrumentation-providers.test.ts +++ b/packages/eve/src/harness/instrumentation-providers.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import { turnIdempotencyKey } from "#harness/instrumentation-lifecycle.js"; import { finalizeInstrumentationProviders, getInstrumentationProviders, @@ -162,6 +163,7 @@ describe("finalizeInstrumentationProviders", () => { const runtime = finalizeInstrumentationProviders({ serviceName: "weather-agent" }); await runtime.hooks.publish({ + idempotencyKey: turnIdempotencyKey("session-1", "turn-1"), rootSessionId: "session-1", sequence: 0, sessionId: "session-1", diff --git a/packages/eve/src/harness/tool-loop.ts b/packages/eve/src/harness/tool-loop.ts index f6c77d8205..ac027e7e19 100644 --- a/packages/eve/src/harness/tool-loop.ts +++ b/packages/eve/src/harness/tool-loop.ts @@ -119,6 +119,7 @@ import { type InstrumentationActionSource, } from "#harness/instrumentation-native-events.js"; import type { InstrumentationAttemptScope } from "#harness/instrumentation-lifecycle.js"; +import { attemptIdempotencyKey } from "#harness/instrumentation-lifecycle.js"; import { resolveParentLineage } from "#harness/parent-lineage.js"; import { ChannelKey } from "#runtime/sessions/runtime-context-keys.js"; import { @@ -1184,6 +1185,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { const result = await executeModelCall(); if (attemptScope !== undefined) { await instrumentationHooks?.publish({ + idempotencyKey: attemptIdempotencyKey(attemptScope), scope: attemptScope, type: "step.attempt.completed", }); @@ -1193,6 +1195,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { if (attemptScope !== undefined) { await instrumentationHooks?.publish({ error, + idempotencyKey: attemptIdempotencyKey(attemptScope), scope: attemptScope, type: "step.attempt.failed", }); diff --git a/packages/eve/src/public/instrumentation/provider.ts b/packages/eve/src/public/instrumentation/provider.ts index fd603f0981..6ef575efbb 100644 --- a/packages/eve/src/public/instrumentation/provider.ts +++ b/packages/eve/src/public/instrumentation/provider.ts @@ -22,9 +22,13 @@ export type { InstrumentationModelRef, InstrumentationOperationRef, InstrumentationParentLineage, + InstrumentationSessionFailedEvent, + InstrumentationSessionSettledEvent, InstrumentationSessionStartedEvent, InstrumentationSessionTransitionEvent, InstrumentationStepAttemptMetadataEvent, + InstrumentationStepAttemptCompletedEvent, + InstrumentationStepAttemptFailedEvent, InstrumentationStepAttemptStartedEvent, InstrumentationStepAttemptTerminalEvent, InstrumentationToolCallCompletedEvent, @@ -32,6 +36,8 @@ export type { InstrumentationToolCallStartedEvent, InstrumentationToolOutput, InstrumentationTraceContext, + InstrumentationTurnFailedEvent, + InstrumentationTurnSettledEvent, InstrumentationTurnStartedEvent, InstrumentationTurnTerminalEvent, InstrumentationUsage, diff --git a/packages/eve/src/tracing/agent-otel-provider.test.ts b/packages/eve/src/tracing/agent-otel-provider.test.ts index 04d2bdfd21..ecd5d469d7 100644 --- a/packages/eve/src/tracing/agent-otel-provider.test.ts +++ b/packages/eve/src/tracing/agent-otel-provider.test.ts @@ -15,7 +15,10 @@ import { SESSION_WINDOW_TURN_LIMIT, } from "#tracing/agent-trace-state.js"; import { + attemptIdempotencyKey, createInstrumentationHooks, + sessionIdempotencyKey, + turnIdempotencyKey, type InstrumentationAttemptScope, type InstrumentationContextRunner, type InstrumentationHooks, @@ -168,19 +171,26 @@ async function emitAttempt(input: { if (input.providerMetadata !== undefined) { await input.hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), providerMetadata: input.providerMetadata, scope, type: "step.attempt.metadata", }); } - await input.hooks.publish({ scope, type: "step.attempt.completed" }); await input.hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + scope, + type: "step.attempt.completed", + }); + await input.hooks.publish({ + idempotencyKey: turnIdempotencyKey(input.sessionId, input.turnId), sessionId: input.sessionId, turnId: input.turnId, type: "turn.completed", }); await input.hooks.publish({ + idempotencyKey: sessionIdempotencyKey(input.sessionId), sessionId: input.sessionId, turnId: input.turnId, type: "session.waiting", @@ -200,12 +210,14 @@ async function publishTurnStarted(input: { await input.hooks.publish({ agentName: "weather", channelKind: "http", + idempotencyKey: sessionIdempotencyKey(input.sessionId), parentTraceContext: input.parentTraceContext, rootSessionId, sessionId: input.sessionId, type: "session.started", }); await input.hooks.publish({ + idempotencyKey: turnIdempotencyKey(input.sessionId, input.turnId), parentLineage: input.parentLineage, parentTraceContext: input.parentTraceContext, rootSessionId, @@ -222,8 +234,18 @@ async function completeTurn( sessionId: string, turnId: string, ): Promise { - await hooks.publish({ sessionId, turnId, type: "turn.completed" }); - await hooks.publish({ sessionId, turnId, type: "session.waiting" }); + await hooks.publish({ + idempotencyKey: turnIdempotencyKey(sessionId, turnId), + sessionId, + turnId, + type: "turn.completed", + }); + await hooks.publish({ + idempotencyKey: sessionIdempotencyKey(sessionId), + sessionId, + turnId, + type: "session.waiting", + }); } function byName(spans: readonly ReadableSpan[], name: string): ReadableSpan[] { @@ -392,11 +414,13 @@ describe("createAgentOtelInstrumentation", () => { await runtime.hooks.publish({ agentName: "weather", channelKind: "http", + idempotencyKey: sessionIdempotencyKey("session-1"), rootSessionId: "session-1", sessionId: "session-1", type: "session.started", }); await runtime.hooks.publish({ + idempotencyKey: turnIdempotencyKey("session-1", "turn-1"), rootSessionId: "session-1", sequence: 0, sessionId: "session-1", diff --git a/packages/eve/src/tracing/agent-otel-provider.ts b/packages/eve/src/tracing/agent-otel-provider.ts index f0e76412b0..e723625776 100644 --- a/packages/eve/src/tracing/agent-otel-provider.ts +++ b/packages/eve/src/tracing/agent-otel-provider.ts @@ -41,6 +41,7 @@ import type { InstrumentationTurnTerminalEvent, InstrumentationUsage, } from "#harness/instrumentation-lifecycle.js"; +import { sessionIdempotencyKey } from "#harness/instrumentation-lifecycle.js"; interface SpanState { readonly context: Context; @@ -104,6 +105,7 @@ export function createAgentOtelInstrumentation( await ensureSessionContext({ agentName: undefined, channelKind: undefined, + idempotencyKey: sessionIdempotencyKey(event.sessionId), parentTraceContext: event.parentTraceContext, rootSessionId: event.rootSessionId, sessionId: event.sessionId, @@ -201,7 +203,10 @@ export function createAgentOtelInstrumentation( if (turn === undefined) return; await input.stateStore.setTurn(event.sessionId, event.turnId, { ...turn, - terminal: { error: event.error, type: event.type }, + terminal: + event.type === "turn.failed" + ? { error: event.error, type: event.type } + : { type: event.type }, }); }; @@ -280,13 +285,13 @@ export function createAgentOtelInstrumentation( if (system !== undefined) span.setAttribute("ai.prompt.system", system); } const state = { context: trace.setSpan(attempt.operation.context, span), span }; - getExecutionContexts(event.scope).models.set(event.id, state.context); - getSpanStates(modelSpans, event.scope).set(event.id, state); + getExecutionContexts(event.scope).models.set(event.idempotencyKey, state.context); + getSpanStates(modelSpans, event.scope).set(event.idempotencyKey, state); }; const onModelCallTerminal = (event: InstrumentationModelCallTerminalEvent): void => { - executionContexts.get(event.scope)?.models.delete(event.id); - const state = takeSpanState(modelSpans, event.scope, event.id); + executionContexts.get(event.scope)?.models.delete(event.idempotencyKey); + const state = takeSpanState(modelSpans, event.scope, event.idempotencyKey); if (state === undefined) return; if (event.type === "model.call.failed") { recordError(state.span, event.error); @@ -379,13 +384,13 @@ export function createAgentOtelInstrumentation( span: actionSpan, toolSpan, }; - getExecutionContexts(event.scope).tools.set(event.id, state.context); - getSpanStates(toolSpans, event.scope).set(event.id, state); + getExecutionContexts(event.scope).tools.set(event.idempotencyKey, state.context); + getSpanStates(toolSpans, event.scope).set(event.idempotencyKey, state); }; const onToolCallTerminal = (event: InstrumentationToolCallTerminalEvent): void => { - executionContexts.get(event.scope)?.tools.delete(event.id); - const state = takeSpanState(toolSpans, event.scope, event.id); + executionContexts.get(event.scope)?.tools.delete(event.idempotencyKey); + const state = takeSpanState(toolSpans, event.scope, event.idempotencyKey); if (state === undefined) return; if (event.type === "tool.call.failed") { recordError(state.toolSpan, event.error); @@ -516,8 +521,8 @@ export function createAgentOtelInstrumentation( const contexts = executionContexts.get(operation.scope); const parent = operation.type === "model.call" - ? contexts?.models.get(operation.id) - : contexts?.tools.get(operation.id); + ? contexts?.models.get(operation.idempotencyKey) + : contexts?.tools.get(operation.idempotencyKey); return parent === undefined ? execute() : context.with(parent, execute); }, }; diff --git a/packages/eve/src/tracing/agent-trace-context-store.test.ts b/packages/eve/src/tracing/agent-trace-context-store.test.ts index 9357f1446e..51c36eb09b 100644 --- a/packages/eve/src/tracing/agent-trace-context-store.test.ts +++ b/packages/eve/src/tracing/agent-trace-context-store.test.ts @@ -53,7 +53,9 @@ describe("ContextAgentTraceStateStore", () => { parentSpanId: "2".repeat(16), startTimeMs: 1_700_000_000_000, }); - expect(store.getTurn("session-1", "turn-1")?.terminal?.error).toMatchObject({ + const terminal = store.getTurn("session-1", "turn-1")?.terminal; + expect(terminal?.type).toBe("turn.failed"); + expect(terminal?.type === "turn.failed" ? terminal.error : undefined).toMatchObject({ message: "failed", }); }); diff --git a/packages/eve/src/tracing/agent-trace-context-store.ts b/packages/eve/src/tracing/agent-trace-context-store.ts index 16fb85431c..3c6111044a 100644 --- a/packages/eve/src/tracing/agent-trace-context-store.ts +++ b/packages/eve/src/tracing/agent-trace-context-store.ts @@ -112,10 +112,9 @@ function serializeState(state: AgentTraceContextState): unknown { terminal: value.terminal === undefined ? undefined - : { - error: serializeError(value.terminal.error), - type: value.terminal.type, - }, + : value.terminal.type === "turn.failed" + ? { error: serializeError(value.terminal.error), type: value.terminal.type } + : { type: value.terminal.type }, }, ]), ), @@ -187,7 +186,7 @@ function deserializeTerminal(value: unknown): AgentTurnTraceState["terminal"] { if (!isRecord(value) || typeof value.type !== "string") return undefined; const type = value.type; if (!isTurnTerminalType(type)) return undefined; - return { error: deserializeError(value.error), type }; + return type === "turn.failed" ? { error: deserializeError(value.error), type } : { type }; } function serializeSpanContext(context: SpanContext): Record { diff --git a/packages/eve/src/tracing/agent-trace-state.ts b/packages/eve/src/tracing/agent-trace-state.ts index dc571f48dd..93e9f20b6a 100644 --- a/packages/eve/src/tracing/agent-trace-state.ts +++ b/packages/eve/src/tracing/agent-trace-state.ts @@ -2,7 +2,8 @@ import type { SpanContext } from "#compiled/@opentelemetry/api/index.js"; import type { InstrumentationParentLineage, - InstrumentationTurnTerminalEvent, + InstrumentationTurnFailedEvent, + InstrumentationTurnSettledEvent, } from "#harness/instrumentation-lifecycle.js"; /** Sized so an ordinary session stays one trace and only an outsized one rolls. */ @@ -24,10 +25,9 @@ export interface AgentTurnTraceState { readonly rootSessionId: string; readonly sequence: number; readonly startTimeMs: number; - readonly terminal?: { - readonly error?: unknown; - readonly type: InstrumentationTurnTerminalEvent["type"]; - }; + readonly terminal?: + | { readonly error: unknown; readonly type: InstrumentationTurnFailedEvent["type"] } + | { readonly type: InstrumentationTurnSettledEvent["type"] }; } /** Provider-owned serializable storage for durable agent trace 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 4cd20e0740..e93c27f689 100644 --- a/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts +++ b/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts @@ -10,6 +10,11 @@ import { ContextContainer, contextStorage } from "#context/container.js"; 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 { + attemptIdempotencyKey, + sessionIdempotencyKey, + turnIdempotencyKey, +} from "#harness/instrumentation-lifecycle.js"; import { installLocalInstrumentationRuntime } from "#tracing/local-instrumentation-runtime.js"; import { LocalTraceSpanProcessor } from "#tracing/local-trace-span-processor.js"; @@ -43,11 +48,13 @@ describe("local instrumentation runtime", () => { await contextStorage.run(new ContextContainer(), async () => { await runtime.hooks.publish({ agentName: "weather", + idempotencyKey: sessionIdempotencyKey("session-1"), rootSessionId: "session-1", sessionId: "session-1", type: "session.started", }); await runtime.hooks.publish({ + idempotencyKey: turnIdempotencyKey("session-1", "turn-1"), rootSessionId: "session-1", sequence: 0, sessionId: "session-1", @@ -109,14 +116,20 @@ describe("local instrumentation runtime", () => { toolOutput: { output: { temperature: 72 }, type: "tool-result" }, }, ]); - await runtime.hooks.publish({ scope, type: "step.attempt.completed" }); await runtime.hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + scope, + type: "step.attempt.completed", + }); + await runtime.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({ + idempotencyKey: sessionIdempotencyKey("session-1"), sessionId: "session-1", turnId: "turn-1", type: "session.waiting",