From ca89d7d8d7a5277d63ccc2d988a914f847fbe5e7 Mon Sep 17 00:00:00 2001 From: Chad Hietala Date: Sat, 8 Aug 2026 09:12:02 -0400 Subject: [PATCH] feat(eve): harden the instrumentation bus and wire providers to it Signed-off-by: Chad Hietala --- .changeset/silver-donkeys-shake.md | 9 + packages/eve/src/evals/cli/eval.ts | 9 + packages/eve/src/execution/node-step.ts | 3 + packages/eve/src/execution/workflow-steps.ts | 11 +- .../src/harness/ai-sdk-hook-bridge.test.ts | 201 ++++---- .../eve/src/harness/ai-sdk-hook-bridge.ts | 112 +++-- packages/eve/src/harness/execute-tool.ts | 1 + .../harness/instrumentation-lifecycle.test.ts | 437 ++++++++++++++++++ .../src/harness/instrumentation-lifecycle.ts | 356 ++++++++++++-- .../instrumentation-native-events.test.ts | 164 ++++--- .../harness/instrumentation-native-events.ts | 116 +++-- .../harness/instrumentation-providers.test.ts | 70 ++- .../src/harness/instrumentation-providers.ts | 46 +- .../harness/instrumentation-setup-context.ts | 7 +- .../src/harness/instrumentation-state.test.ts | 140 ++++++ .../eve/src/harness/instrumentation-state.ts | 342 ++++++++++++++ .../eve/src/harness/runtime-actions.test.ts | 48 ++ packages/eve/src/harness/runtime-actions.ts | 11 + packages/eve/src/harness/tool-loop.test.ts | 43 +- packages/eve/src/harness/tool-loop.ts | 22 +- .../application/compiled-artifacts.ts | 1 + .../internal/application/dev-environment.ts | 22 + .../src/public/instrumentation/provider.ts | 74 ++- .../tracing/agent-action-instrumentation.ts | 155 +++++++ .../src/tracing/agent-otel-provider.test.ts | 179 ++++++- .../eve/src/tracing/agent-otel-provider.ts | 100 ++-- .../tracing/agent-trace-context-store.test.ts | 4 +- .../src/tracing/agent-trace-context-store.ts | 100 +++- packages/eve/src/tracing/agent-trace-state.ts | 64 ++- .../install-instrumentation-runtime.test.ts | 2 +- .../install-instrumentation-runtime.ts | 12 +- ...l-instrumentation-runtime.scenario.test.ts | 32 +- .../otel-registration.scenario.test.ts | 8 +- .../eval-command-environment.scenario.test.ts | 4 + 34 files changed, 2462 insertions(+), 443 deletions(-) create mode 100644 .changeset/silver-donkeys-shake.md create mode 100644 packages/eve/src/harness/instrumentation-lifecycle.test.ts create mode 100644 packages/eve/src/harness/instrumentation-state.test.ts create mode 100644 packages/eve/src/harness/instrumentation-state.ts create mode 100644 packages/eve/src/tracing/agent-action-instrumentation.ts diff --git a/.changeset/silver-donkeys-shake.md b/.changeset/silver-donkeys-shake.md new file mode 100644 index 0000000000..a95bd41678 --- /dev/null +++ b/.changeset/silver-donkeys-shake.md @@ -0,0 +1,9 @@ +--- +"eve": patch +--- + +Harden the instrumentation bus behind `experimental.instrumentationProviders`. +Events carry replay-stable identity, ordinary tools expose both durable +`action.*` and AI SDK `tool.call.*` boundaries, and provider state, abandonment, +flush, and shutdown survive the worker lifecycle. Provider setup also receives +the local eval run reference through `context.evaluation`. diff --git a/packages/eve/src/evals/cli/eval.ts b/packages/eve/src/evals/cli/eval.ts index 26a7090752..2ed4991ac0 100644 --- a/packages/eve/src/evals/cli/eval.ts +++ b/packages/eve/src/evals/cli/eval.ts @@ -1,8 +1,13 @@ +import { randomUUID } from "node:crypto"; import { readFile } from "node:fs/promises"; import { basename, join } from "node:path"; import { loadDevelopmentEnvironmentFiles } from "#cli/dev/environment.js"; import { shutdownActiveSandboxHandles } from "#execution/sandbox/active-handles.js"; +import { + EVE_EVALUATION_ENV_FLAG, + EVE_EVALUATION_RUN_ID_ENV, +} from "#internal/application/dev-environment.js"; import { resolveApplicationRoot } from "#internal/application/paths.js"; import { createDevelopmentServer, type DevelopmentServer } from "#internal/nitro/host.js"; import { createEvalClient } from "#evals/cli/eval-client.js"; @@ -141,6 +146,10 @@ export async function runEvalCommand( url: options.url, }); } else { + // Set before the server boots, because a provider's `setup` reads it + // once at startup and never again. + process.env[EVE_EVALUATION_ENV_FLAG] = "1"; + process.env[EVE_EVALUATION_RUN_ID_ENV] = randomUUID(); devServer = createDevelopmentServer(appRoot, { host: "127.0.0.1", port: 0 }); const started = await devServer.start(); client = await createEvalClient({ kind: "local", url: started.url }); diff --git a/packages/eve/src/execution/node-step.ts b/packages/eve/src/execution/node-step.ts index 5d23e429b8..e27cabb084 100644 --- a/packages/eve/src/execution/node-step.ts +++ b/packages/eve/src/execution/node-step.ts @@ -4,6 +4,7 @@ import type { Runtime, SessionCapabilities } from "#channel/types.js"; import { dispatchDynamicModelEvent } from "#context/dynamic-model-lifecycle.js"; import { createHarnessDelegationToolDefinition } from "#execution/delegation-tool.js"; 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 { resolveInstalledPackageInfo } from "#internal/application/package.js"; @@ -260,6 +261,8 @@ function resolveHarnessToolDefinition(input: { rawExecute, scope: def.name, }), + frameworkAction: + isFrameworkTool && def.name === LOAD_SKILL_TOOL_NAME ? "load-skill" : undefined, inputSchema: def.inputSchema ?? UNSPECIFIED_INPUT_SCHEMA, name: def.name, approval: def.approval, diff --git a/packages/eve/src/execution/workflow-steps.ts b/packages/eve/src/execution/workflow-steps.ts index 4d9ee6590f..45d3f9547d 100644 --- a/packages/eve/src/execution/workflow-steps.ts +++ b/packages/eve/src/execution/workflow-steps.ts @@ -24,6 +24,7 @@ import { BundleKey, ChannelKey } from "#runtime/sessions/runtime-context-keys.js import { runStep } from "#context/run-step.js"; import { deserializeContext, serializeContext } from "#context/serialize.js"; import { getHarnessEmissionState } from "#harness/emission.js"; +import { preserveSerializedInstrumentationState } from "#harness/instrumentation-state.js"; import { preserveSerializedAgentTraceState } from "#tracing/agent-trace-context-store.js"; import { readTurnSleepDurationMs } from "#harness/turn-sleep.js"; import { isTurnCancellation, throwIfTurnAborted } from "#harness/turn-cancellation.js"; @@ -452,11 +453,15 @@ export async function turnStep(rawInput: TurnStepInput): Promise { 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))}`, + ); }, }, + name, }; }; const hooks = createInstrumentationHooks([provider("a"), provider("b")]); @@ -55,7 +60,7 @@ describe("createAiSdkHookBridge", () => { }, ]); - const id = `${scope.attemptId}:model:call-1:0`; + const id = modelCallIdempotencyKey(scope, 0); expect(calls).toEqual([ `a:started:${id}`, `b:started:${id}`, @@ -91,10 +96,13 @@ 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) }, + name: "recorder", + }, ]); const bridge = createAiSdkHookBridge(scope, hooks, (operation, execute) => { - ids.push(operation.id); + ids.push(operation.idempotencyKey); return execute(); }); Reflect.apply(bridge.onStart!, bridge, [ @@ -108,7 +116,7 @@ describe("createAiSdkHookBridge", () => { await bridge.executeLanguageModelCall!({ callId: "call-1", execute: async () => "result" }); - const expected = `${scope.attemptId}:model:call-1:0`; + const expected = modelCallIdempotencyKey(scope, 0); expect(ids).toEqual([expected, expected]); }); @@ -126,6 +134,26 @@ 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) }, + name: "keys", + }, + ]); + + 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([modelCallIdempotencyKey(scope, 2), modelCallIdempotencyKey(scope, 2)]); + }); + it("publishes step provider metadata as step.metadata, skipping steps without any", async () => { const events: InstrumentationStepAttemptMetadataEvent[] = []; const hooks = createInstrumentationHooks([ @@ -135,6 +163,7 @@ describe("createAiSdkHookBridge", () => { events.push(event); }, }, + name: "metadata", }, ]); const bridge = createAiSdkHookBridge(scope, hooks); @@ -146,6 +175,7 @@ describe("createAiSdkHookBridge", () => { expect(events).toEqual([ { + idempotencyKey: attemptIdempotencyKey(scope), providerMetadata: { gateway: { cost: "0.000082" } }, scope, type: "step.attempt.metadata", @@ -162,11 +192,13 @@ describe("createAiSdkHookBridge", () => { throw new Error("provider failed"); }, }, + name: "thrower", }, { events: { "model.call.completed": after, }, + name: "after", }, ]); const bridge = createAiSdkHookBridge(scope, hooks); @@ -190,7 +222,9 @@ describe("createAiSdkHookBridge", () => { it("terminalizes started operations when the attempt errors", async () => { const after = vi.fn(); - const hooks = createInstrumentationHooks([{ events: { "model.call.failed": after } }]); + const hooks = createInstrumentationHooks([ + { events: { "model.call.failed": after }, name: "after" }, + ]); const bridge = createAiSdkHookBridge(scope, hooks); await Reflect.apply(bridge.onLanguageModelCallStart!, bridge, [ @@ -201,6 +235,7 @@ describe("createAiSdkHookBridge", () => { expect(after).toHaveBeenCalledExactlyOnceWith( expect.objectContaining({ error, type: "model.call.failed" }), + expect.anything(), ); }); @@ -214,8 +249,8 @@ describe("createAiSdkHookBridge", () => { }); const started = vi.fn(); const hooks = createInstrumentationHooks([ - { events: { "step.attempt.started": mutator } }, - { events: { "step.attempt.started": started } }, + { events: { "step.attempt.started": mutator }, name: "mutator" }, + { events: { "step.attempt.started": started }, name: "started" }, ]); const bridge = createAiSdkHookBridge(scope, hooks); @@ -225,12 +260,13 @@ describe("createAiSdkHookBridge", () => { await Reflect.apply(bridge.onStepStart!, bridge, [{ callId: "call-1", stepNumber: 0 }]); const expected = { + idempotencyKey: attemptIdempotencyKey(scope), operation: { modelId: "model", operationId: "ai.streamText", provider: "test" }, scope, type: "step.attempt.started", }; - expect(mutator).toHaveBeenCalledExactlyOnceWith(expected); - expect(started).toHaveBeenCalledExactlyOnceWith(expected); + expect(mutator).toHaveBeenCalledExactlyOnceWith(expected, expect.anything()); + expect(started).toHaveBeenCalledExactlyOnceWith(expected, expect.anything()); }); it("projects the model call callbacks onto eve fields only", async () => { @@ -249,7 +285,7 @@ describe("createAiSdkHookBridge", () => { expect(Object.isFrozen(event.usage.inputTokenDetails)).toBe(true); }); const hooks = createInstrumentationHooks([ - { events: { "model.call.completed": after, "model.call.started": before } }, + { events: { "model.call.completed": after, "model.call.started": before }, name: "spy" }, ]); const bridge = createAiSdkHookBridge(scope, hooks); @@ -285,33 +321,39 @@ describe("createAiSdkHookBridge", () => { }, ]); - expect(before).toHaveBeenCalledExactlyOnceWith({ - id: `${scope.attemptId}:model:call-1:0`, - input: { instructions: "be brief", messages: [{ content: "hi", role: "user" }] }, - model: { modelId: "model", provider: "test" }, - scope, - type: "model.call.started", - }); + expect(before).toHaveBeenCalledExactlyOnceWith( + { + idempotencyKey: modelCallIdempotencyKey(scope, 0), + input: { instructions: "be brief", messages: [{ content: "hi", role: "user" }] }, + model: { modelId: "model", provider: "test" }, + scope, + type: "model.call.started", + }, + expect.anything(), + ); // An unrecognized part kind is dropped rather than forwarded, so widening // InstrumentationContentPart is what makes a new kind reachable. - expect(after).toHaveBeenCalledExactlyOnceWith({ - content: [ - { text: "thinking", type: "reasoning" }, - { text: "hello", type: "text" }, - { input: { a: 1 }, toolName: "search", type: "tool-call" }, - { input: { a: 1 }, output: "ok", toolName: "search", type: "tool-result" }, - { error: "boom", input: { a: 2 }, toolName: "search", type: "tool-error" }, - ], - finishReason: "tool-calls", - id: `${scope.attemptId}:model:call-1:0`, - scope, - type: "model.call.completed", - usage: { - inputTokenDetails: { cacheReadTokens: 3, cacheWriteTokens: 4 }, - inputTokens: 1, - outputTokens: 2, + expect(after).toHaveBeenCalledExactlyOnceWith( + { + content: [ + { text: "thinking", type: "reasoning" }, + { text: "hello", type: "text" }, + { input: { a: 1 }, toolName: "search", type: "tool-call" }, + { input: { a: 1 }, output: "ok", toolName: "search", type: "tool-result" }, + { error: "boom", input: { a: 2 }, toolName: "search", type: "tool-error" }, + ], + finishReason: "tool-calls", + idempotencyKey: modelCallIdempotencyKey(scope, 0), + scope, + type: "model.call.completed", + usage: { + inputTokenDetails: { cacheReadTokens: 3, cacheWriteTokens: 4 }, + inputTokens: 1, + outputTokens: 2, + }, }, - }); + expect.anything(), + ); }); it.each([ @@ -334,8 +376,16 @@ describe("createAiSdkHookBridge", () => { expect(Object.isFrozen(event)).toBe(true); expect(Object.isFrozen(event.output)).toBe(true); }); + const actionStarted = vi.fn(); const hooks = createInstrumentationHooks([ - { events: { "tool.call.completed": after, "tool.call.started": before } }, + { + events: { + "action.started": actionStarted, + "tool.call.completed": after, + "tool.call.started": before, + }, + name: "spy", + }, ]); const bridge = createAiSdkHookBridge(scope, hooks); const toolCall = { input: { q: "eve" }, toolCallId: "tool-1", toolName: "search" }; @@ -345,43 +395,30 @@ describe("createAiSdkHookBridge", () => { { callId: "call-1", toolCall, toolExecutionMs: 1, toolOutput }, ]); - expect(before).toHaveBeenCalledExactlyOnceWith({ - callId: "tool-1", - id: `${scope.attemptId}:tool:tool-1:0`, - input: { q: "eve" }, - kind: "tool-call", - scope, - toolName: "search", - type: "tool.call.started", - }); - expect(after).toHaveBeenCalledExactlyOnceWith({ - id: `${scope.attemptId}:tool:tool-1:0`, - output: expected, - scope, - type: "tool.call.completed", - }); + expect(before).toHaveBeenCalledExactlyOnceWith( + { + callId: "tool-1", + idempotencyKey: `tool:${scope.attemptId}:tool-1:0`, + input: { q: "eve" }, + scope, + toolName: "search", + type: "tool.call.started", + }, + expect.anything(), + ); + expect(after).toHaveBeenCalledExactlyOnceWith( + { + idempotencyKey: `tool:${scope.attemptId}:tool-1:0`, + output: expected, + scope, + type: "tool.call.completed", + }, + expect.anything(), + ); + expect(actionStarted).not.toHaveBeenCalled(); }, ); - it("labels a tool call with the kind the harness resolves", async () => { - const started = vi.fn(); - const hooks = createInstrumentationHooks([{ events: { "tool.call.started": started } }]); - const bridge = createAiSdkHookBridge(scope, hooks, undefined, (toolName) => - toolName === "research" ? "subagent-call" : "tool-call", - ); - - for (const toolName of ["research", "search"]) { - await Reflect.apply(bridge.onToolExecutionStart!, bridge, [ - { callId: `call-${toolName}`, toolCall: { input: {}, toolCallId: toolName, toolName } }, - ]); - } - - expect(started.mock.calls.map(([event]) => [event.toolName, event.kind])).toEqual([ - ["research", "subagent-call"], - ["search", "tool-call"], - ]); - }); - it("keeps each provider's state to itself", async () => { const observed = new Map(); const provider = (name: string): InstrumentationProviderDefinition => { @@ -389,10 +426,11 @@ 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`), }, + name, }; }; const hooks = createInstrumentationHooks([provider("a"), provider("b")]); @@ -422,7 +460,9 @@ describe("createAiSdkHookBridge", () => { it("skips a terminal handler when the operation never started", async () => { const completed = vi.fn(); - const hooks = createInstrumentationHooks([{ events: { "model.call.completed": completed } }]); + const hooks = createInstrumentationHooks([ + { events: { "model.call.completed": completed }, name: "completed" }, + ]); const bridge = createAiSdkHookBridge(scope, hooks); // No onLanguageModelCallStart, so the bridge holds no id and publishes @@ -450,16 +490,17 @@ 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)); }, }, + name: "parallel", }, ]); const bridge = createAiSdkHookBridge(scope, hooks); @@ -474,8 +515,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..ce6a39d586 100644 --- a/packages/eve/src/harness/ai-sdk-hook-bridge.ts +++ b/packages/eve/src/harness/ai-sdk-hook-bridge.ts @@ -1,7 +1,6 @@ import type { Telemetry } from "ai"; import type { - InstrumentationActionKind, InstrumentationAttemptScope, InstrumentationStepAttemptStartedEvent, InstrumentationContentPart, @@ -15,20 +14,18 @@ import type { InstrumentationToolOutput, InstrumentationUsage, } from "#harness/instrumentation-lifecycle.js"; +import { + attemptIdempotencyKey, + modelCallIdempotencyKey, + toolCallIdempotencyKey, +} from "#harness/instrumentation-lifecycle.js"; type TelemetryEvent = Parameters>[0]; -/** - * Reports what eve dispatches one tool name as. The AI SDK only knows the - * name, so the kind has to come back from the harness. - */ -export type ActionKindResolver = (toolName: string) => InstrumentationActionKind; - interface AttemptState { - readonly modelIds: Map; - readonly resolveActionKind: ActionKindResolver; + readonly modelKeys: Map; 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; @@ -39,13 +36,11 @@ export function createAiSdkHookBridge( scope: InstrumentationAttemptScope, hooks: InstrumentationHooks, runInContext: InstrumentationContextRunner = directRunInContext, - resolveActionKind: ActionKindResolver = defaultResolveActionKind, ): Telemetry { const state: AttemptState = { - modelIds: new Map(), - resolveActionKind, + modelKeys: new Map(), scope, - toolIds: new Map(), + toolKeys: new Map(), }; return { @@ -62,22 +57,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 +82,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 +90,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,44 +123,43 @@ 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); } } const directRunInContext: InstrumentationContextRunner = (_operation, execute) => execute(); -const defaultResolveActionKind: ActionKindResolver = () => "tool-call"; - function toStepAttemptStarted( state: AttemptState, ): 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 +172,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,20 +239,15 @@ 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, toolName: source.toolCall.toolName, type: "tool.call.started", @@ -260,11 +256,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/execute-tool.ts b/packages/eve/src/harness/execute-tool.ts index ad92b038ec..cca9701f70 100644 --- a/packages/eve/src/harness/execute-tool.ts +++ b/packages/eve/src/harness/execute-tool.ts @@ -23,6 +23,7 @@ export interface HarnessToolDefinition { readonly approvalKey?: (toolInput: Readonly>) => string; readonly description: string; readonly execute?: (input: any, options: ToolExecuteOptions) => any; + readonly frameworkAction?: "load-skill"; readonly inputSchema: FlexibleSchema; readonly name: string; readonly approval?: Approval; 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..e4dd151517 --- /dev/null +++ b/packages/eve/src/harness/instrumentation-lifecycle.test.ts @@ -0,0 +1,437 @@ +import { describe, expect, it, vi } from "vitest"; + +import { ContextContainer, contextStorage } from "#context/container.js"; +import { deserializeContext, serializeContext } from "#context/serialize.js"; +import { + attemptIdempotencyKey, + actionIdempotencyKey, + createInstrumentationHooks, + modelCallIdempotencyKey, + sessionIdempotencyKey, + toolCallIdempotencyKey, + turnIdempotencyKey, + type InstrumentationAttemptScope, +} from "#harness/instrumentation-lifecycle.js"; +import { + findInstrumentationActionScopeForCall, + instrumentationStateSlot, + rememberInstrumentationActionScope, +} from "#harness/instrumentation-state.js"; + +const scope: InstrumentationAttemptScope = { + attemptId: "session-1:turn-1:0:0", + attemptIndex: 0, + sessionId: "session-1", + stepIndex: 0, + turnId: "turn-1", +}; + +const turnKey = (turnId: string): string => turnIdempotencyKey("session-1", turnId); + +describe("idempotency keys", () => { + it("separates classes that share an identifier", () => { + // A provider using the key as a row id would otherwise merge a session and + // a turn whose generated ids happened to match. + const shared = "shared-1"; + expect(sessionIdempotencyKey(shared)).not.toBe(turnIdempotencyKey(shared, shared)); + }); + + it("separates operation classes", () => { + expect(actionIdempotencyKey(scope.sessionId, scope.turnId, "call-1")).not.toBe( + modelCallIdempotencyKey(scope, 0), + ); + expect(modelCallIdempotencyKey(scope, 0)).not.toBe(toolCallIdempotencyKey(scope, "call-1", 0)); + }); + + it("separates identical turn IDs in different sessions", () => { + expect(turnIdempotencyKey("session-1", "turn_0")).not.toBe( + turnIdempotencyKey("session-2", "turn_0"), + ); + }); + + it("separates two retries of one step", () => { + const retried = { ...scope, attemptId: "session-1:turn-1:0:1", attemptIndex: 1 }; + expect(attemptIdempotencyKey(scope)).not.toBe(attemptIdempotencyKey(retried)); + }); + + it("derives the same key from the same identity", () => { + // The whole promise: a later process observing the same operation writes + // the same row rather than a duplicate. + expect(actionIdempotencyKey(scope.sessionId, scope.turnId, "call-1")).toBe( + actionIdempotencyKey(scope.sessionId, scope.turnId, "call-1"), + ); + }); +}); + +describe("handler timeout", () => { + const started = (turnId: string) => + ({ + idempotencyKey: turnKey(turnId), + rootSessionId: "session-1", + sequence: 0, + sessionId: "session-1", + turnId, + type: "turn.started", + }) as const; + + const completed = (turnId: string) => + ({ + idempotencyKey: turnKey(turnId), + sessionId: "session-1", + turnId, + type: "turn.completed", + }) as const; + + const hang = () => new Promise(() => {}); + + it("lets the providers behind a hanging one run", async () => { + const after = vi.fn(); + const hooks = createInstrumentationHooks( + [ + { events: { "turn.started": hang }, name: "hangs" }, + { events: { "turn.started": after }, name: "after" }, + ], + { handlerTimeoutMs: 1 }, + ); + + await contextStorage.run(new ContextContainer(), () => hooks.publish(started("turn-1"))); + + expect(after).toHaveBeenCalledOnce(); + }); + + it("skips the terminal of an operation it abandoned", async () => { + // The abandoned handler is still running, so it may never have opened the + // operation its terminal would close. + const abandonedTerminal = vi.fn(); + const healthyTerminal = vi.fn(); + const startHooks = createInstrumentationHooks( + [ + { events: { "turn.completed": abandonedTerminal, "turn.started": hang }, name: "hangs" }, + { events: { "turn.completed": healthyTerminal }, name: "healthy" }, + ], + { handlerTimeoutMs: 1 }, + ); + + const startContext = new ContextContainer(); + await contextStorage.run(startContext, () => startHooks.publish(started("turn-1"))); + const terminalContext = await deserializeContext(await serializeContext(startContext)); + await contextStorage.run(terminalContext, async () => { + const terminalHooks = createInstrumentationHooks( + [ + { events: { "turn.completed": abandonedTerminal }, name: "hangs" }, + { events: { "turn.completed": healthyTerminal }, name: "healthy" }, + ], + { handlerTimeoutMs: 1 }, + ); + await terminalHooks.publish(completed("turn-1")); + }); + + expect(abandonedTerminal).not.toHaveBeenCalled(); + expect(healthyTerminal).toHaveBeenCalledOnce(); + }); + + it("abandons one operation, not the provider", async () => { + const terminal = vi.fn(); + const hooks = createInstrumentationHooks( + [ + { + events: { + "turn.completed": terminal, + "turn.started": (event) => (event.turnId === "turn-1" ? hang() : Promise.resolve()), + }, + name: "hangs-once", + }, + ], + { handlerTimeoutMs: 1 }, + ); + + await contextStorage.run(new ContextContainer(), async () => { + await hooks.publish(started("turn-1")); + await hooks.publish(started("turn-2")); + await hooks.publish(completed("turn-1")); + await hooks.publish(completed("turn-2")); + }); + + expect(terminal).toHaveBeenCalledOnce(); + expect(terminal.mock.calls[0]?.[0]).toMatchObject({ turnId: "turn-2" }); + }); + + it("releases the state of an operation it abandoned", async () => { + // The abandoned handler never sees its terminal, so it could not release + // what it staged even if it wanted to. + const context = new ContextContainer(); + const hooks = createInstrumentationHooks( + [ + { + events: { + "turn.started": (_event, ctx) => { + ctx.state.set("open"); + return hang(); + }, + }, + name: "hangs", + }, + ], + { handlerTimeoutMs: 1 }, + ); + + await contextStorage.run(context, async () => { + await hooks.publish(started("turn-1")); + await hooks.publish(completed("turn-1")); + expect(instrumentationStateSlot("hangs", turnKey("turn-1")).get()).toBeUndefined(); + }); + }); + + it("ignores a timed-out handler that writes after its terminal", async () => { + let resume!: () => void; + const continued = new Promise((resolve) => { + resume = resolve; + }); + const context = new ContextContainer(); + const hooks = createInstrumentationHooks( + [ + { + events: { + "turn.started": async (_event, ctx) => { + await continued; + ctx.state.set("too-late"); + }, + }, + name: "slow", + }, + ], + { handlerTimeoutMs: 1 }, + ); + + await contextStorage.run(context, async () => { + await hooks.publish(started("turn-1")); + await hooks.publish(completed("turn-1")); + resume(); + await continued; + await Promise.resolve(); + expect(instrumentationStateSlot("slow", turnKey("turn-1")).get()).toBeUndefined(); + }); + }); + + it("does not abandon an operation when a point-event handler times out", async () => { + const completed = vi.fn(); + const hooks = createInstrumentationHooks( + [ + { + events: { + "step.attempt.completed": completed, + "step.attempt.metadata": hang, + }, + name: "slow-metadata", + }, + ], + { handlerTimeoutMs: 1 }, + ); + + await contextStorage.run(new ContextContainer(), async () => { + await hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + providerMetadata: {}, + scope, + type: "step.attempt.metadata", + }); + await hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + scope, + type: "step.attempt.completed", + }); + }); + + expect(completed).toHaveBeenCalledOnce(); + }); +}); + +describe("handler state", () => { + const started = (turnId: string) => + ({ + idempotencyKey: turnKey(turnId), + rootSessionId: "session-1", + sequence: 0, + sessionId: "session-1", + turnId, + type: "turn.started", + }) as const; + + const completed = (turnId: string) => + ({ + idempotencyKey: turnKey(turnId), + sessionId: "session-1", + turnId, + type: "turn.completed", + }) as const; + + it("carries a value from a start to its terminal", async () => { + const closed = vi.fn(); + const hooks = createInstrumentationHooks([ + { + events: { + "turn.completed": (_event, ctx) => closed(ctx.state.get()), + "turn.started": (_event, ctx) => ctx.state.set({ rowId: "row-1" }), + }, + name: "sink", + }, + ]); + + await contextStorage.run(new ContextContainer(), async () => { + await hooks.publish(started("turn-1")); + await hooks.publish(completed("turn-1")); + }); + + expect(closed).toHaveBeenCalledWith({ rowId: "row-1" }); + }); + + it("releases the slot once the terminal has run", async () => { + const context = new ContextContainer(); + const hooks = createInstrumentationHooks([ + { + events: { + "turn.completed": () => {}, + "turn.started": (_event, ctx) => ctx.state.set("open"), + }, + name: "sink", + }, + ]); + + await contextStorage.run(context, async () => { + await hooks.publish(started("turn-1")); + await hooks.publish(completed("turn-1")); + expect(instrumentationStateSlot("sink", turnKey("turn-1")).get()).toBeUndefined(); + }); + }); + + it("releases the slot of a provider with no terminal handler", async () => { + // Nothing will ever read it, and the provider has no handler in which to + // notice the operation ended. + const context = new ContextContainer(); + const hooks = createInstrumentationHooks([ + { events: { "turn.started": (_event, ctx) => ctx.state.set("open") }, name: "sink" }, + ]); + + await contextStorage.run(context, async () => { + await hooks.publish(started("turn-1")); + await hooks.publish(completed("turn-1")); + expect(instrumentationStateSlot("sink", turnKey("turn-1")).get()).toBeUndefined(); + }); + }); + + it("keeps one provider out of another's slot", async () => { + const read = vi.fn(); + const hooks = createInstrumentationHooks([ + { events: { "turn.started": (_event, ctx) => ctx.state.set("mine") }, name: "writer" }, + { events: { "turn.started": (_event, ctx) => read(ctx.state.get()) }, name: "reader" }, + ]); + + await contextStorage.run(new ContextContainer(), async () => { + await hooks.publish(started("turn-1")); + }); + + expect(read).toHaveBeenCalledWith(undefined); + }); + + it("releases unterminated model state when its attempt ends", async () => { + const modelKey = modelCallIdempotencyKey(scope, 0); + const hooks = createInstrumentationHooks([ + { + events: { + "model.call.started": (_event, ctx) => ctx.state.set("open"), + }, + name: "sink", + }, + ]); + + await contextStorage.run(new ContextContainer(), async () => { + await hooks.publish({ + idempotencyKey: modelKey, + input: { messages: [] }, + model: { modelId: "model", provider: "test" }, + scope, + type: "model.call.started", + }); + expect(instrumentationStateSlot("sink", modelKey).get()).toBe("open"); + + await hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + scope, + type: "step.attempt.completed", + }); + + expect(instrumentationStateSlot("sink", modelKey).get()).toBeUndefined(); + }); + }); + + it("does not sweep durable action state with the originating attempt", async () => { + const actionKey = actionIdempotencyKey(scope.sessionId, scope.turnId, "call-1"); + const hooks = createInstrumentationHooks([ + { + events: { + "action.started": (_event, ctx) => ctx.state.set("open"), + }, + name: "sink", + }, + ]); + + await contextStorage.run(new ContextContainer(), async () => { + await hooks.publish({ + callId: "call-1", + idempotencyKey: actionKey, + input: {}, + kind: "tool-call", + name: "tool", + scope, + type: "action.started", + }); + await hooks.publish({ + idempotencyKey: attemptIdempotencyKey(scope), + scope, + type: "step.attempt.completed", + }); + + expect(instrumentationStateSlot("sink", actionKey).get()).toBe("open"); + }); + }); + + it("sweeps durable action state when its turn is cancelled", async () => { + const actionKey = actionIdempotencyKey(scope.sessionId, scope.turnId, "call-1"); + const failed = vi.fn(); + const hooks = createInstrumentationHooks([ + { + events: { + "action.failed": failed, + "action.started": (_event, ctx) => ctx.state.set("open"), + }, + name: "sink", + }, + ]); + + await contextStorage.run(new ContextContainer(), async () => { + rememberInstrumentationActionScope(actionKey, scope); + await hooks.publish({ + callId: "call-1", + idempotencyKey: actionKey, + input: {}, + kind: "tool-call", + name: "tool", + scope, + type: "action.started", + }); + await hooks.publish({ + idempotencyKey: turnKey(scope.turnId), + sessionId: scope.sessionId, + turnId: scope.turnId, + type: "turn.cancelled", + }); + + expect(instrumentationStateSlot("sink", actionKey).get()).toBeUndefined(); + expect(findInstrumentationActionScopeForCall(scope.sessionId, "call-1")).toBeUndefined(); + }); + expect(failed).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ idempotencyKey: actionKey, type: "action.failed" }), + expect.anything(), + ); + }); +}); diff --git a/packages/eve/src/harness/instrumentation-lifecycle.ts b/packages/eve/src/harness/instrumentation-lifecycle.ts index 78c501ac7b..1dd3ec5478 100644 --- a/packages/eve/src/harness/instrumentation-lifecycle.ts +++ b/packages/eve/src/harness/instrumentation-lifecycle.ts @@ -1,3 +1,14 @@ +import { + abandonInstrumentationState, + instrumentationStateSlot, + isInstrumentationStateAbandoned, + releaseInstrumentationAttemptState, + releaseInstrumentationTurnState, + takeInstrumentationActionScopes, + type InstrumentationStateOwner, + type InstrumentationStateSlot, + releaseInstrumentationState, +} from "#harness/instrumentation-state.js"; import { createLogger, formatError } from "#internal/logging.js"; /** @@ -70,19 +81,66 @@ export type InstrumentationContentPart = }; /** - * What eve dispatched a tool call as. The model sees every action as a tool, - * so this is the only thing that separates a subagent or remote-agent call - * from an ordinary tool in a trace. + * What eve dispatched an action as. The model sees every action as a tool, so + * this is the only thing that separates a subagent or remote-agent call from an + * ordinary tool in a trace. */ -export type InstrumentationActionKind = "remote-agent-call" | "subagent-call" | "tool-call"; - -/** How one tool execution ended. */ -export type InstrumentationToolOutput = +export type InstrumentationActionKind = + | "load-skill" + | "remote-agent-call" + | "subagent-call" + | "tool-call"; + +/** How one action ended. */ +export type InstrumentationActionOutput = | { readonly type: "result"; readonly output: unknown } | { readonly type: "error"; readonly error: unknown }; +/** + * Every event carries an `idempotencyKey` naming the operation it is about: a + * start and its terminal share one, and two operations never collide. + * + * Every part is identity eve reconstructs on replay — session and turn ids, + * `scope.attemptId` (itself `session:turn:step:attempt`), AI SDK step number, + * and durable runtime-action call ids. A provider writing rows can use the key + * as its row id and be idempotent by construction. + */ +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)}`; +} + +/** Runtime action call IDs are durable and unique within one session. */ +export function actionIdempotencyKey(sessionId: string, turnId: string, callId: string): string { + return `action:${sessionId}:${turnId}:${callId}`; +} + export interface InstrumentationStepAttemptStartedEvent { readonly type: "step.attempt.started"; + readonly idempotencyKey: string; readonly operation: InstrumentationOperationRef; readonly scope: InstrumentationAttemptScope; } @@ -91,6 +149,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 +181,7 @@ export interface InstrumentationParentLineage { */ export interface InstrumentationSessionSettledEvent { readonly type: "session.completed" | "session.waiting"; + readonly idempotencyKey: string; readonly sessionId: string; readonly turnId?: string; } @@ -129,6 +189,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 +200,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 +209,49 @@ export interface InstrumentationTurnStartedEvent { readonly turnId: string; } -export interface InstrumentationTurnTerminalEvent { - readonly type: "turn.cancelled" | "turn.completed" | "turn.failed"; - readonly error?: unknown; +/** + * A turn that ended without a failure. + * + * `turn.cancelled` sits here rather than with the failed shape because + * cancellation is not an error: the harness settles a cancelled turn as + * `turn.cancelled` → `session.waiting`, with no failure surfaced anywhere. + */ +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 +259,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 +276,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 +284,7 @@ export interface InstrumentationModelCallCompletedEvent { export interface InstrumentationModelCallFailedEvent { readonly type: "model.call.failed"; readonly error: unknown; - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; } @@ -199,19 +292,20 @@ export type InstrumentationModelCallTerminalEvent = | InstrumentationModelCallCompletedEvent | InstrumentationModelCallFailedEvent; +export type InstrumentationToolOutput = InstrumentationActionOutput; + 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; readonly toolName: string; } export interface InstrumentationToolCallCompletedEvent { readonly type: "tool.call.completed"; - readonly id: string; + readonly idempotencyKey: string; readonly output: InstrumentationToolOutput; readonly scope: InstrumentationAttemptScope; } @@ -219,7 +313,7 @@ export interface InstrumentationToolCallCompletedEvent { export interface InstrumentationToolCallFailedEvent { readonly type: "tool.call.failed"; readonly error: unknown; - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; } @@ -227,19 +321,63 @@ export type InstrumentationToolCallTerminalEvent = | InstrumentationToolCallCompletedEvent | InstrumentationToolCallFailedEvent; +/** + * One thing the agent did on the model's behalf. `kind` is what separates a + * subagent or remote-agent call from an ordinary tool; `name` is the name the + * model called, which is the tool name for every kind. + */ +export interface InstrumentationActionStartedEvent { + readonly type: "action.started"; + readonly callId: string; + readonly idempotencyKey: string; + readonly input: unknown; + readonly kind: InstrumentationActionKind; + readonly name: string; + readonly scope: InstrumentationAttemptScope; +} + +export interface InstrumentationActionCompletedEvent { + readonly type: "action.completed"; + readonly idempotencyKey: string; + readonly output: InstrumentationActionOutput; + readonly scope: InstrumentationAttemptScope; +} + +export interface InstrumentationActionFailedEvent { + readonly type: "action.failed"; + readonly error: unknown; + readonly idempotencyKey: string; + readonly scope: InstrumentationAttemptScope; +} + +export type InstrumentationActionTerminalEvent = + | InstrumentationActionCompletedEvent + | InstrumentationActionFailedEvent; + +/** The second argument to every handler. */ +export interface InstrumentationHandlerContext { + /** Durable state scoped to this provider and this operation. */ + readonly state: InstrumentationStateSlot; +} + /** * The AI SDK can omit a model terminal when an incomplete stream closes. A - * provider that correlates starts with terminals must scope that state to the - * attempt and release anything still open when the step attempt terminates. + * handler can use `ctx.state` for durable correlation when a terminal arrives, + * but providers must scope live resources to the attempt and release anything + * still open when the step attempt terminates. */ -export type InstrumentationEventHandler = (event: TEvent) => void | PromiseLike; +export type InstrumentationEventHandler = ( + event: TEvent, + ctx: InstrumentationHandlerContext, +) => void | PromiseLike; /** Internal provider shape mirrored by the future public hook contract. */ export interface InstrumentationProviderDefinition { + readonly name: string; 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; @@ -248,21 +386,27 @@ export interface InstrumentationProviderDefinition { readonly "session.failed"?: InstrumentationEventHandler; readonly "session.started"?: InstrumentationEventHandler; readonly "session.waiting"?: InstrumentationEventHandler; + readonly "action.started"?: InstrumentationEventHandler; + readonly "action.completed"?: InstrumentationEventHandler; + readonly "action.failed"?: InstrumentationEventHandler; 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; }; + /** Drains anything buffered. Driven by the runtime, not by the bus. */ readonly flush?: () => void | PromiseLike; - readonly name?: string; + /** Releases resources when the process is going away. */ 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 = + | InstrumentationActionStartedEvent + | InstrumentationActionTerminalEvent | InstrumentationModelCallStartedEvent | InstrumentationModelCallTerminalEvent | InstrumentationToolCallStartedEvent @@ -288,14 +432,14 @@ 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 type: "tool.call"; } | { - readonly id: string; + readonly idempotencyKey: string; readonly scope: InstrumentationAttemptScope; - readonly type: "tool.call"; + readonly type: "model.call"; }; /** Provider-neutral hook operations consumed by the AI SDK bridge. */ @@ -305,24 +449,164 @@ export interface InstrumentationHooks { const log = createLogger("harness.instrumentation-lifecycle"); +/** + * Dispatch is sequential and awaited, so a handler that never settles stalls + * every provider behind it and the agent turn with them. + */ +const DEFAULT_HANDLER_TIMEOUT_MS = 5_000; + +export interface CreateInstrumentationHooksOptions { + readonly handlerTimeoutMs?: number; +} + /** Creates failure-isolated hooks backed by an ordered provider list. */ export function createInstrumentationHooks( providers: readonly InstrumentationProviderDefinition[], + options: CreateInstrumentationHooksOptions = {}, ): InstrumentationHooks { + const handlerTimeoutMs = options.handlerTimeoutMs ?? DEFAULT_HANDLER_TIMEOUT_MS; + const publish = async (event: InstrumentationEvent): Promise => { + const terminal = isTerminal(event.type); + const startedBoundary = event.type.endsWith(".started"); + const attemptTerminal = + event.type === "step.attempt.completed" || event.type === "step.attempt.failed"; + const owner = stateOwner(event); + const cleanupSession = event.type === "session.completed" || event.type === "session.failed"; + const cleanupTurn = event.type === "turn.cancelled" || event.type === "turn.failed"; + + if (cleanupSession || cleanupTurn) { + const pendingActions = takeInstrumentationActionScopes( + event.sessionId, + cleanupTurn ? event.turnId : undefined, + ); + const error = terminalActionError(event); + for (const action of pendingActions) { + await publish({ + error, + idempotencyKey: action.idempotencyKey, + scope: action.scope, + type: "action.failed", + }); + } + } + for (const provider of providers) { + // The operation is over for this provider either way, so release what it + // staged at the start. Nothing downstream can read it now, and a provider + // that was abandoned or has no terminal handler could never release it + // itself. + const release = (): void => { + if (terminal) releaseInstrumentationState(provider.name, event.idempotencyKey); + if (attemptTerminal) + releaseInstrumentationAttemptState(provider.name, event.scope.attemptId); + if (cleanupSession) releaseInstrumentationTurnState(provider.name, event.sessionId); + if (cleanupTurn) + releaseInstrumentationTurnState(provider.name, event.sessionId, event.turnId); + }; + + if (isInstrumentationStateAbandoned(provider.name, event.idempotencyKey)) { + release(); + continue; + } + const handler = provider.events?.[event.type]; - if (handler === undefined) continue; + if (handler === undefined) { + release(); + continue; + } + + const state = instrumentationStateSlot(provider.name, event.idempotencyKey, owner); + const ctx: InstrumentationHandlerContext = { state }; + try { - await (handler as InstrumentationEventHandler)(event); + const settled = await withTimeout( + () => (handler as InstrumentationEventHandler)(event, ctx), + handlerTimeoutMs, + () => { + state.revoke(); + if (startedBoundary) { + abandonInstrumentationState(provider.name, event.idempotencyKey, owner); + } + }, + ); + // The handler cannot be cancelled, only left running. Handing it a + // terminal now would complete an operation it may never have started, + // so the rest of this operation is not its to see. + if (!settled && startedBoundary) { + log.warn("instrumentation provider timed out", { + boundary: event.type, + provider: provider.name, + timeoutMs: handlerTimeoutMs, + }); + } } catch (error) { log.warn("instrumentation provider failed", { boundary: event.type, error: formatError(error), + provider: provider.name, }); + } finally { + state.revoke(); + release(); } } }; return { publish }; } + +/** Resolves false when the deadline wins; rejects with whatever the handler threw. */ +async function withTimeout( + run: () => void | PromiseLike, + timeoutMs: number, + onTimeout: () => void, +): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + Promise.resolve(run()).then(() => true), + new Promise((resolve) => { + timer = setTimeout(() => { + onTimeout(); + resolve(false); + }, timeoutMs); + }), + ]); + } finally { + clearTimeout(timer); + } +} + +/** Model and SDK tool children are scoped to an attempt; runtime actions are not. */ +function stateOwner(event: InstrumentationEvent): InstrumentationStateOwner { + if (!("scope" in event)) return {}; + if (event.type.startsWith("action.")) { + return { sessionId: event.scope.sessionId, turnId: event.scope.turnId }; + } + return event.type.startsWith("model.call.") || + event.type.startsWith("tool.call.") || + event.type.startsWith("step.attempt.") + ? { attemptId: event.scope.attemptId } + : {}; +} + +function terminalActionError( + event: + | InstrumentationSessionFailedEvent + | InstrumentationSessionSettledEvent + | InstrumentationTurnFailedEvent + | InstrumentationTurnSettledEvent, +): unknown { + if (event.type === "session.failed" || event.type === "turn.failed") return event.error; + return new Error( + event.type === "turn.cancelled" + ? "The action was cancelled with its turn." + : "The session completed before the action settled.", + ); +} + +/** The vocabulary spells every terminal transition as one of these suffixes. */ +function isTerminal(type: InstrumentationEvent["type"]): boolean { + return type.endsWith(".completed") || type.endsWith(".failed") || type.endsWith(".cancelled"); +} diff --git a/packages/eve/src/harness/instrumentation-native-events.test.ts b/packages/eve/src/harness/instrumentation-native-events.test.ts index a67b8f24f1..3581e7a883 100644 --- a/packages/eve/src/harness/instrumentation-native-events.test.ts +++ b/packages/eve/src/harness/instrumentation-native-events.test.ts @@ -1,7 +1,9 @@ -import { jsonSchema } from "ai"; import { describe, expect, it } from "vitest"; +import { ContextContainer, contextStorage } from "#context/container.js"; +import { deserializeContext, serializeContext } from "#context/serialize.js"; import { + createActionResultEvent, createActionsRequestedEvent, createSessionStartedEvent, createSessionWaitingEvent, @@ -12,6 +14,11 @@ import { } from "#protocol/message.js"; import { createInstrumentationHandleEvent } from "#harness/instrumentation-native-events.js"; import type { InstrumentationHooks } from "#harness/instrumentation-lifecycle.js"; +import { + actionIdempotencyKey, + sessionIdempotencyKey, + turnIdempotencyKey, +} from "#harness/instrumentation-lifecycle.js"; describe("createInstrumentationHandleEvent", () => { it("publishes native lifecycle transitions after durable handling", async () => { @@ -84,7 +91,14 @@ describe("createInstrumentationHandleEvent", () => { await handleEvent(createSessionWaitingEvent()); - expect(events).toEqual([{ sessionId: "session-1", turnId: "turn-1", type: "session.waiting" }]); + expect(events).toEqual([ + { + idempotencyKey: sessionIdempotencyKey("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 +126,7 @@ describe("createInstrumentationHandleEvent", () => { expect(events.filter((event) => event.type === "turn.started")).toEqual([ { + idempotencyKey: turnIdempotencyKey("child-1", "child-turn-1"), parentLineage, parentTraceContext: undefined, rootSessionId: "session-1", @@ -121,6 +136,7 @@ describe("createInstrumentationHandleEvent", () => { type: "turn.started", }, { + idempotencyKey: turnIdempotencyKey("child-1", "child-turn-2"), parentLineage, parentTraceContext: undefined, rootSessionId: "session-1", @@ -132,7 +148,7 @@ describe("createInstrumentationHandleEvent", () => { ]); }); - it("publishes each non-executable delegation once from actions.requested", async () => { + it("publishes every runtime action and settles it in a replacement worker", async () => { const events: unknown[] = []; const scope = { attemptId: "session-1:turn-1:0:0", @@ -141,50 +157,7 @@ describe("createInstrumentationHandleEvent", () => { stepIndex: 0, turnId: "turn-1", }; - const tools = new Map([ - [ - "delegate", - { - description: "Delegate work.", - inputSchema: jsonSchema({ type: "object" }), - name: "delegate", - runtimeAction: { - kind: "subagent-call" as const, - nodeId: "workers", - subagentName: "worker", - }, - }, - ], - [ - "add", - { - description: "Add numbers.", - execute: () => 3, - inputSchema: jsonSchema({ type: "object" }), - name: "add", - }, - ], - [ - "remote", - { - description: "Call a remote agent.", - inputSchema: jsonSchema({ type: "object" }), - name: "remote", - runtimeAction: { - kind: "remote-agent-call" as const, - nodeId: "remote-agents", - remoteAgentName: "analyst", - subagentName: "analyst", - }, - }, - ], - ]); - const handleEvent = createInstrumentationHandleEvent({ - getActionSource: () => ({ scope, tools }), - handleEvent: async () => {}, - hooks: { publish: async (event) => void events.push(event) }, - sessionId: "session-1", - })!; + const context = new ContextContainer(); const requested = createActionsRequestedEvent({ actions: [ { @@ -196,6 +169,7 @@ describe("createInstrumentationHandleEvent", () => { nodeId: "workers", subagentName: "worker", }, + { callId: "skill-1", input: { name: "research" }, kind: "load-skill" }, { callId: "remote-1", description: "Call a remote agent.", @@ -212,30 +186,104 @@ describe("createInstrumentationHandleEvent", () => { turnId: "turn-1", }); - await handleEvent(requested); - await handleEvent(requested); + await contextStorage.run(context, async () => { + const handleEvent = createInstrumentationHandleEvent({ + getAttemptScope: () => scope, + handleEvent: async () => {}, + hooks: { publish: async (event) => void events.push(event) }, + sessionId: "session-1", + })!; + await handleEvent(requested); + await handleEvent(requested); + }); - expect(events).toEqual([ + const restored = await deserializeContext(await serializeContext(context)); + await contextStorage.run(restored, async () => { + const handleEvent = createInstrumentationHandleEvent({ + handleEvent: async () => {}, + hooks: { publish: async (event) => void events.push(event) }, + sessionId: "session-1", + })!; + await handleEvent( + createActionResultEvent({ + result: { + callId: "delegate-1", + kind: "subagent-result", + origin: "dispatch", + output: "unavailable", + isError: true, + subagentName: "worker", + }, + sequence: 0, + stepIndex: 0, + turnId: "turn-2", + }), + ); + await handleEvent( + createActionResultEvent({ + result: { + callId: "add-1", + kind: "tool-result", + output: 3, + toolName: "add", + }, + sequence: 0, + stepIndex: 0, + turnId: "turn-2", + }), + ); + }); + + expect(events.slice(0, 4)).toEqual([ { callId: "delegate-1", - id: "session-1:turn-1:0:0:tool:delegate-1:0", + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "delegate-1"), input: { task: "research" }, kind: "subagent-call", + name: "delegate", + scope, + type: "action.started", + }, + { + callId: "skill-1", + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "skill-1"), + input: { name: "research" }, + kind: "load-skill", + name: "load_skill", scope, - toolName: "delegate", - type: "tool.call.started", + type: "action.started", }, { callId: "remote-1", - id: "session-1:turn-1:0:0:tool:remote-1:0", + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "remote-1"), input: { task: "analyze" }, kind: "remote-agent-call", + name: "remote", scope, - toolName: "remote", - type: "tool.call.started", + type: "action.started", + }, + { + callId: "add-1", + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "add-1"), + input: { a: 1, b: 2 }, + kind: "tool-call", + name: "add", + scope, + type: "action.started", }, ]); - expect(Object.isFrozen(events[0])).toBe(true); - expect(Object.isFrozen(events[1])).toBe(true); + expect(events[4]).toMatchObject({ + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "delegate-1"), + scope, + type: "action.failed", + }); + expect(events[5]).toEqual({ + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "add-1"), + output: { output: 3, type: "result" }, + scope, + type: "action.completed", + }); + expect(events).toHaveLength(6); + expect(events.every(Object.isFrozen)).toBe(true); }); }); diff --git a/packages/eve/src/harness/instrumentation-native-events.ts b/packages/eve/src/harness/instrumentation-native-events.ts index 42daaaab96..63a06ec480 100644 --- a/packages/eve/src/harness/instrumentation-native-events.ts +++ b/packages/eve/src/harness/instrumentation-native-events.ts @@ -1,22 +1,28 @@ import type { UnstampedMessageStreamEvent } from "#protocol/message.js"; import type { + InstrumentationActionFailedEvent, + InstrumentationActionStartedEvent, InstrumentationAttemptScope, InstrumentationHooks, InstrumentationParentLineage, InstrumentationPointEvent, - InstrumentationToolCallStartedEvent, InstrumentationTraceContext, } from "#harness/instrumentation-lifecycle.js"; -import type { HandleEventFn, HarnessToolMap } from "#harness/types.js"; - -export interface InstrumentationActionSource { - readonly scope: InstrumentationAttemptScope; - readonly tools: HarnessToolMap; -} +import { + actionIdempotencyKey, + sessionIdempotencyKey, + turnIdempotencyKey, +} from "#harness/instrumentation-lifecycle.js"; +import { + rememberInstrumentationActionScope, + takeInstrumentationActionScopeForCall, +} from "#harness/instrumentation-state.js"; +import type { HandleEventFn } from "#harness/types.js"; +import type { RuntimeActionRequest } from "#runtime/actions/types.js"; export interface CreateInstrumentationHandleEventInput { readonly agentName?: string; - readonly getActionSource?: () => InstrumentationActionSource | undefined; + readonly getAttemptScope?: () => InstrumentationAttemptScope | undefined; readonly handleEvent?: HandleEventFn; readonly hooks?: InstrumentationHooks; readonly parentLineage?: InstrumentationParentLineage; @@ -43,39 +49,83 @@ export function createInstrumentationHandleEvent( if (event.type === "turn.started") activeTurnId = event.data.turnId; if (lifecycleEvent !== undefined) await hooks.publish(lifecycleEvent); if (event.type === "actions.requested") { - await publishDelegationActions(event, input, hooks, publishedActions); + await publishActionStarts(event, input, hooks, publishedActions); + } else if (event.type === "action.result") { + await publishActionTerminal(event, input, hooks); } }; } -async function publishDelegationActions( +async function publishActionStarts( event: Extract, input: CreateInstrumentationHandleEventInput, hooks: InstrumentationHooks, published: Set, ): Promise { - const source = input.getActionSource?.(); - if (source === undefined) return; + const scope = input.getAttemptScope?.(); + if (scope === undefined) return; for (const action of event.data.actions) { - if (action.kind !== "subagent-call" && action.kind !== "remote-agent-call") continue; - const tool = source.tools.get(action.name); - if (tool?.runtimeAction === undefined || tool.execute !== undefined) continue; - const deduplicationKey = `${source.scope.attemptId}:${action.callId}`; - if (published.has(deduplicationKey)) continue; - published.add(deduplicationKey); + const idempotencyKey = actionIdempotencyKey(input.sessionId, event.data.turnId, action.callId); + if (published.has(idempotencyKey)) continue; + published.add(idempotencyKey); + rememberInstrumentationActionScope(idempotencyKey, scope); await hooks.publish( Object.freeze({ callId: action.callId, - id: `${source.scope.attemptId}:tool:${action.callId}:0`, + idempotencyKey, input: action.input, - kind: tool.runtimeAction.kind, - scope: source.scope, - toolName: tool.name, - type: "tool.call.started", - } satisfies InstrumentationToolCallStartedEvent), + kind: action.kind, + name: actionName(action), + scope, + type: "action.started", + } satisfies InstrumentationActionStartedEvent), + ); + } +} + +async function publishActionTerminal( + event: Extract, + input: CreateInstrumentationHandleEventInput, + hooks: InstrumentationHooks, +): Promise { + const correlation = takeInstrumentationActionScopeForCall( + input.sessionId, + event.data.result.callId, + ); + if (correlation === undefined) return; + const { idempotencyKey, scope } = correlation; + + if (event.data.status === "completed") { + await hooks.publish( + Object.freeze({ + idempotencyKey, + output: Object.freeze({ output: event.data.result.output, type: "result" }), + scope, + type: "action.completed", + }), ); + return; } + + const error = + event.data.error === undefined + ? event.data.result.output + : Object.assign(new Error(event.data.error.message), { code: event.data.error.code }); + await hooks.publish( + Object.freeze({ + error, + idempotencyKey, + scope, + type: "action.failed", + } satisfies InstrumentationActionFailedEvent), + ); +} + +function actionName(action: RuntimeActionRequest): string { + if (action.kind === "tool-call") return action.toolName; + if (action.kind === "load-skill") return "load_skill"; + return action.name; } function toLifecycleEvent( @@ -87,6 +137,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 +145,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 +172,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..5b7f847034 100644 --- a/packages/eve/src/harness/instrumentation-providers.test.ts +++ b/packages/eve/src/harness/instrumentation-providers.test.ts @@ -1,5 +1,10 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import { turnIdempotencyKey } from "#harness/instrumentation-lifecycle.js"; +import { + EVE_EVALUATION_ENV_FLAG, + EVE_EVALUATION_RUN_ID_ENV, +} from "#internal/application/dev-environment.js"; import { finalizeInstrumentationProviders, getInstrumentationProviders, @@ -58,6 +63,7 @@ describe("registerInstrumentationProvider", () => { expect(contexts).toHaveLength(1); expect(contexts[0]?.agentName).toBe("weather-agent"); expect(contexts[0]?.environment).toMatch(/^(development|preview|production)$/); + expect(contexts[0]?.evaluation).toBeUndefined(); expect(contexts[0]?.frameworkVersion).toEqual(expect.any(String)); }); @@ -73,6 +79,23 @@ describe("registerInstrumentationProvider", () => { expect(contexts[0]?.environment).toBe("preview"); }); + it("reports an evaluation server to setup", async () => { + vi.stubEnv(EVE_EVALUATION_ENV_FLAG, "1"); + vi.stubEnv(EVE_EVALUATION_RUN_ID_ENV, "eval-run-1"); + + const contexts: ProviderSetupContext[] = []; + await register( + "otel", + defineInstrumentation({ + setup: (context) => { + contexts.push(context); + }, + }), + ); + + expect(contexts[0]?.evaluation).toEqual({ runId: "eval-run-1" }); + }); + it("registers nothing for a disabled slot", async () => { await register("local", disableInstrumentation()); @@ -150,29 +173,58 @@ describe("seedInstrumentationProviders", () => { }); describe("finalizeInstrumentationProviders", () => { + const turnStarted = { + idempotencyKey: turnIdempotencyKey("session-1", "turn-1"), + rootSessionId: "session-1", + sequence: 0, + sessionId: "session-1", + turnId: "turn-1", + type: "turn.started", + } as const; + beforeEach(() => { vi.unstubAllEnvs(); delete (globalThis as Record)[REGISTRY_GLOBAL_KEY]; delete (globalThis as Record)[RUNTIME_GLOBAL_KEY]; }); - it("installs a bus for authored providers without an OpenTelemetry destination", async () => { + it("publishes to an authored handler", async () => { const started = vi.fn(); await register("rows", defineInstrumentation({ events: { "turn.started": started } })); const runtime = finalizeInstrumentationProviders({ serviceName: "weather-agent" }); - await runtime.hooks.publish({ - rootSessionId: "session-1", - sequence: 0, - sessionId: "session-1", - turnId: "turn-1", - type: "turn.started", - }); + await runtime.hooks.publish(turnStarted); expect(started).toHaveBeenCalledOnce(); + expect(started.mock.calls[0]?.[0]).toMatchObject({ turnId: "turn-1" }); + }); + + it("still runs execution when no destination was declared", async () => { + // A directory with no `otel()` has nothing to hang a span on, so + // `runInContext` degrades to running the work directly rather than + // going missing. + await register("rows", defineInstrumentation({})); + + const runtime = finalizeInstrumentationProviders({ serviceName: "weather-agent" }); + const result = await runtime.runInContext( + { + idempotencyKey: "tool:session-1:turn-1:0:0:call-1:0", + scope: { + attemptId: "session-1:turn-1:0:0", + attemptIndex: 0, + sessionId: "session-1", + stepIndex: 0, + turnId: "turn-1", + }, + type: "tool.call", + }, + () => Promise.resolve("ran"), + ); + + expect(result).toBe("ran"); }); - it("drives authored flush and shutdown hooks", async () => { + it("drains and releases every provider", async () => { const flush = vi.fn(); const shutdown = vi.fn(); await register("rows", defineInstrumentation({ flush, shutdown })); diff --git a/packages/eve/src/harness/instrumentation-providers.ts b/packages/eve/src/harness/instrumentation-providers.ts index 82254ad43e..b6a2053506 100644 --- a/packages/eve/src/harness/instrumentation-providers.ts +++ b/packages/eve/src/harness/instrumentation-providers.ts @@ -1,7 +1,4 @@ -import type { - InstrumentationEvent, - InstrumentationProviderDefinition, -} from "#harness/instrumentation-lifecycle.js"; +import type { InstrumentationProviderDefinition } from "#harness/instrumentation-lifecycle.js"; import { getInstrumentationRuntime, type InstrumentationRuntime, @@ -15,9 +12,7 @@ import { collectOtelPipeline } from "#tracing/otel-declaration.js"; import { isInstrumentationDisabled, isInstrumentationProvider, - type Handler, type InstrumentationProvider, - type ProviderContext, } from "#public/instrumentation/provider.js"; /** @@ -100,16 +95,17 @@ export function getInstrumentationProviders(): readonly RegisteredInstrumentatio } /** - * Builds the process's OpenTelemetry pipeline from the registered providers. + * Installs the process instrumentation runtime from the registered providers. * * Called once by the generated Nitro plugin after every slot has registered, - * which is also why it cannot happen inside `setup`: the pipeline is the union - * of every destination declared in the directory, so no single file knows - * enough to build it. A `setup` that reaches for a tracer therefore gets the - * no-op one; declare destinations as values and let this assemble them. + * which is also why it cannot happen inside `setup`: the OpenTelemetry pipeline + * is the union of every destination declared in the directory, so no single + * file knows enough to build it. A `setup` that reaches for a tracer therefore + * gets the no-op one; declare destinations as values and let this assemble + * them. * - * A directory that declared no OpenTelemetry at all leaves the global tracer - * provider slot alone. + * A directory that declared no OpenTelemetry at all still gets a bus. Its + * providers see every event; they just have no spans to hang them on. * * @internal — not part of the public API. */ @@ -125,25 +121,29 @@ export function finalizeInstrumentationProviders(input: { }); } -/** Drains and releases the process runtime from Nitro's close hook. */ +/** + * Releases every registered provider and OTel processor from Nitro's close + * hook, the last point a buffered exporter can still reach the network. + */ export async function shutdownInstrumentationProviders(): Promise { await getInstrumentationRuntime()?.shutdown(); } -const PROVIDER_CONTEXT: ProviderContext = Object.freeze({}); - +/** + * Adapts an authored provider onto the internal bus contract. + * + * The event maps are the same shape — the public one is derived from the + * internal union and both handlers take `(event, ctx)` — so only the name has + * to be supplied. + */ function toProviderDefinition( entry: RegisteredInstrumentationProvider, ): InstrumentationProviderDefinition { - const events: Record void | PromiseLike> = {}; - for (const [type, handler] of Object.entries(entry.provider.events ?? {})) { - if (handler === undefined) continue; - const invoke = handler as Handler; - events[type] = (event: InstrumentationEvent) => invoke(event, PROVIDER_CONTEXT); - } return { - events: events as InstrumentationProviderDefinition["events"], + events: entry.provider.events as InstrumentationProviderDefinition["events"], flush: entry.provider.flush, + // The file the provider came from, which is the only name an author can + // recognize in a log line about it. name: entry.slot, shutdown: entry.provider.shutdown, }; diff --git a/packages/eve/src/harness/instrumentation-setup-context.ts b/packages/eve/src/harness/instrumentation-setup-context.ts index 53d0b44935..b23dfe575a 100644 --- a/packages/eve/src/harness/instrumentation-setup-context.ts +++ b/packages/eve/src/harness/instrumentation-setup-context.ts @@ -1,4 +1,7 @@ -import { isEveDevEnvironment } from "#internal/application/dev-environment.js"; +import { + isEveDevEnvironment, + resolveEveEvaluationRunId, +} from "#internal/application/dev-environment.js"; import { resolveInstalledPackageInfo } from "#internal/application/package.js"; import type { ProviderSetupContext } from "#public/instrumentation/provider.js"; @@ -13,9 +16,11 @@ import type { ProviderSetupContext } from "#public/instrumentation/provider.js"; * @internal — not part of the public API. */ export function createInstrumentationSetupContext(agentName: string): ProviderSetupContext { + const evaluationRunId = resolveEveEvaluationRunId(); return { agentName, environment: resolveInstrumentationEnvironment(), + evaluation: evaluationRunId === undefined ? undefined : { runId: evaluationRunId }, frameworkVersion: resolveInstalledPackageInfo().version, }; } diff --git a/packages/eve/src/harness/instrumentation-state.test.ts b/packages/eve/src/harness/instrumentation-state.test.ts new file mode 100644 index 0000000000..76f5d0bd67 --- /dev/null +++ b/packages/eve/src/harness/instrumentation-state.test.ts @@ -0,0 +1,140 @@ +import { describe, expect, it } from "vitest"; + +import { ContextContainer, contextStorage } from "#context/container.js"; +import { deserializeContext, serializeContext } from "#context/serialize.js"; +import { + abandonInstrumentationState, + instrumentationStateSlot, + isInstrumentationStateAbandoned, + preserveSerializedInstrumentationState, + releaseInstrumentationAttemptState, + releaseInstrumentationState, +} from "#harness/instrumentation-state.js"; + +describe("instrumentationStateSlot", () => { + it("survives a step boundary", async () => { + // The point of durable state: an approval-gated action starts in one + // process and completes in another. + const context = new ContextContainer(); + contextStorage.run(context, () => { + instrumentationStateSlot("sink", "action:a").set({ rowId: "row-1" }); + }); + + const restored = await deserializeContext(await serializeContext(context)); + contextStorage.run(restored, () => { + expect(instrumentationStateSlot("sink", "action:a").get()).toEqual({ rowId: "row-1" }); + }); + }); + + it("separates providers and operations", () => { + contextStorage.run(new ContextContainer(), () => { + instrumentationStateSlot("a", "turn:1").set("a-1"); + instrumentationStateSlot("b", "turn:1").set("b-1"); + instrumentationStateSlot("a", "turn:2").set("a-2"); + + expect(instrumentationStateSlot("a", "turn:1").get()).toBe("a-1"); + expect(instrumentationStateSlot("b", "turn:1").get()).toBe("b-1"); + expect(instrumentationStateSlot("a", "turn:2").get()).toBe("a-2"); + }); + }); + + it("releases on set(undefined)", () => { + contextStorage.run(new ContextContainer(), () => { + const slot = instrumentationStateSlot("sink", "turn:1"); + slot.set("held"); + slot.set(undefined); + expect(slot.get()).toBeUndefined(); + }); + }); + + it("rejects a value that cannot survive the step boundary", () => { + contextStorage.run(new ContextContainer(), () => { + // Thrown at the write so the handler that wrote it is named, rather than + // at serialization where the step is blamed instead. + expect(() => instrumentationStateSlot("sink", "turn:1").set(new Date() as never)).toThrow( + TypeError, + ); + }); + }); + + it("ignores writes through a revoked lease", () => { + contextStorage.run(new ContextContainer(), () => { + const lease = instrumentationStateSlot("sink", "turn:1"); + lease.set("before"); + lease.revoke(); + lease.set("after"); + + expect(instrumentationStateSlot("sink", "turn:1").get()).toBe("before"); + expect(lease.get()).toBeUndefined(); + }); + }); + + it("persists abandonment across a step boundary", async () => { + const context = new ContextContainer(); + contextStorage.run(context, () => { + abandonInstrumentationState("sink", "model:1", { attemptId: "attempt-1" }); + }); + + const restored = await deserializeContext(await serializeContext(context)); + contextStorage.run(restored, () => { + expect(isInstrumentationStateAbandoned("sink", "model:1")).toBe(true); + }); + }); + + it("releases only records owned by one attempt", () => { + contextStorage.run(new ContextContainer(), () => { + instrumentationStateSlot("sink", "model:1", { attemptId: "attempt-1" }).set("one"); + instrumentationStateSlot("sink", "model:2", { attemptId: "attempt-2" }).set("two"); + + releaseInstrumentationAttemptState("sink", "attempt-1"); + + expect(instrumentationStateSlot("sink", "model:1").get()).toBeUndefined(); + expect(instrumentationStateSlot("sink", "model:2").get()).toBe("two"); + }); + }); +}); + +describe("releaseInstrumentationState", () => { + it("drops only the named slot", () => { + contextStorage.run(new ContextContainer(), () => { + instrumentationStateSlot("a", "turn:1").set("a-1"); + instrumentationStateSlot("b", "turn:1").set("b-1"); + + releaseInstrumentationState("a", "turn:1"); + + expect(instrumentationStateSlot("a", "turn:1").get()).toBeUndefined(); + expect(instrumentationStateSlot("b", "turn:1").get()).toBe("b-1"); + }); + }); + + it("writes nothing when the provider staged nothing", async () => { + // Every provider is released on every terminal, so the common case must not + // create the durable entry just to delete an absent key. + const context = new ContextContainer(); + contextStorage.run(context, () => { + releaseInstrumentationState("never-wrote", "turn:1"); + }); + + expect(await serializeContext(context)).toEqual({}); + }); +}); + +describe("preserveSerializedInstrumentationState", () => { + it("keeps provider state from a discarded step", () => { + const preserved = preserveSerializedInstrumentationState( + { authored: "original" }, + { authored: "discarded", "eve.harness.instrumentationState": { "sink\0action:a": 1 } }, + ); + + expect(preserved).toEqual({ + authored: "original", + "eve.harness.instrumentationState": { "sink\0action:a": 1 }, + }); + }); + + it("leaves the original alone when the step staged nothing", () => { + expect(preserveSerializedInstrumentationState({ authored: "original" }, {})).toEqual({ + authored: "original", + }); + }); +}); diff --git a/packages/eve/src/harness/instrumentation-state.ts b/packages/eve/src/harness/instrumentation-state.ts new file mode 100644 index 0000000000..79817a7e2c --- /dev/null +++ b/packages/eve/src/harness/instrumentation-state.ts @@ -0,0 +1,342 @@ +import { contextStorage, loadContext } from "#context/container.js"; +import { ContextKey } from "#context/key.js"; +import { type JsonValue, parseJsonValue } from "#shared/json.js"; +import type { InstrumentationAttemptScope } from "#harness/instrumentation-lifecycle.js"; + +/** + * What every provider has staged, flattened into one durable slot. + * + * Flat rather than nested by provider, because every read and write is already + * scoped to a single `(provider, operation)` pair — nesting would buy a grouping + * nothing asks for and make releasing one operation a two-level rewrite. + */ +interface InstrumentationStateRecord { + abandoned?: true; + attemptId?: string; + sessionId?: string; + turnId?: string; + value?: JsonValue; +} + +export interface InstrumentationStateOwner { + readonly attemptId?: string; + readonly sessionId?: string; + readonly turnId?: string; +} + +type InstrumentationStateMap = Readonly>; +type InstrumentationActionScopeMap = Readonly>; + +/** + * Provider state lives in serialized Workflow context, not in the harness, so a + * value staged by `action.started` in one process is still there when + * `action.completed` runs in another. + */ +const InstrumentationStateKey = new ContextKey( + "eve.harness.instrumentationState", + { + codec: { + deserialize: deserializeState, + serialize: (state) => state, + }, + }, +); + +const InstrumentationActionScopeKey = new ContextKey( + "eve.harness.instrumentationActionScopes", + { + codec: { + deserialize: deserializeActionScopes, + serialize: (state) => state, + }, + }, +); + +/** + * Keeps provider state from an interrupted step's context changes. + * + * A cancelled step's context writes are discarded wholesale. Provider state has + * to be an exception for the same reason eve's own trace state is: the + * cancellation epilogue still publishes `turn.cancelled`, and a provider that + * staged something at the start of the operation being cancelled needs it there + * to close cleanly. Without this, the terminal arrives with an empty slot and + * whatever the provider opened is never closed. + */ +export function preserveSerializedInstrumentationState( + original: Record, + interrupted: Record, +): Record { + let preserved = original; + for (const key of [InstrumentationStateKey, InstrumentationActionScopeKey]) { + const state = interrupted[key.name]; + if (state !== undefined) preserved = { ...preserved, [key.name]: state }; + } + return preserved; +} + +/** One provider's view of its own state for one operation. */ +export interface InstrumentationStateSlot { + get(): JsonValue | undefined; + /** Stages a value; `undefined` releases the slot. */ + set(value: JsonValue | undefined): void; +} + +export interface InstrumentationStateLease extends InstrumentationStateSlot { + /** Makes later reads empty and writes no-ops. */ + revoke(): void; +} + +/** + * Scopes state to one provider and one operation. + * + * Two providers handling the same event get separate slots, and the same + * provider gets a separate slot per operation, so neither can read or clobber + * the other's. + */ +export function instrumentationStateSlot( + provider: string, + idempotencyKey: string, + owner: InstrumentationStateOwner = {}, +): InstrumentationStateLease { + const key = stateKey(provider, idempotencyKey); + let active = true; + return { + get: () => + active ? contextStorage.getStore()?.get(InstrumentationStateKey)?.[key]?.value : undefined, + revoke: () => { + active = false; + }, + set: (value) => { + if (!active) return; + // Reject a lossy value here rather than at the step boundary, where the + // throw would be attributed to serialization instead of to the handler + // that wrote it. + const staged = value === undefined ? undefined : parseJsonValue(value); + writeInstrumentationState((state) => { + if (staged === undefined) return writeSlot(state, key, undefined); + const current = state[key]; + const record: InstrumentationStateRecord = { value: staged }; + if (current?.abandoned === true) record.abandoned = true; + assignOwner(record, owner); + return writeSlot(state, key, record); + }); + }, + }; +} + +/** Persists that a provider's start handler timed out for this operation. */ +export function abandonInstrumentationState( + provider: string, + idempotencyKey: string, + owner: InstrumentationStateOwner = {}, +): void { + const key = stateKey(provider, idempotencyKey); + writeInstrumentationState((state) => { + const current = state[key]; + const resolvedOwner = { + attemptId: owner.attemptId ?? current?.attemptId, + sessionId: owner.sessionId ?? current?.sessionId, + turnId: owner.turnId ?? current?.turnId, + }; + const record: InstrumentationStateRecord = { abandoned: true }; + assignOwner(record, resolvedOwner); + if (current?.value !== undefined) record.value = current.value; + return writeSlot(state, key, record); + }); +} + +export function isInstrumentationStateAbandoned(provider: string, idempotencyKey: string): boolean { + return ( + contextStorage.getStore()?.get(InstrumentationStateKey)?.[stateKey(provider, idempotencyKey)] + ?.abandoned === true + ); +} + +/** + * Drops what a provider staged for an operation that has reached its terminal. + * + * eve releases rather than leaving it to the provider: a handler that never + * settles is abandoned and never sees its terminal, so a provider given the job + * would leak exactly the slots it could not know about. + */ +export function releaseInstrumentationState(provider: string, idempotencyKey: string): void { + const key = stateKey(provider, idempotencyKey); + const current = contextStorage.getStore()?.get(InstrumentationStateKey); + // Most providers stage nothing. Writing unconditionally would create the + // durable entry for all of them just to delete a key that was never there. + if (current?.[key] === undefined) return; + writeInstrumentationState((state) => writeSlot(state, key, undefined)); +} + +/** Releases child state whose terminal may be omitted when an attempt ends. */ +export function releaseInstrumentationAttemptState(provider: string, attemptId: string): void { + const prefix = `${provider}\0`; + const current = contextStorage.getStore()?.get(InstrumentationStateKey); + if ( + current === undefined || + !Object.entries(current).some( + ([key, record]) => key.startsWith(prefix) && record.attemptId === attemptId, + ) + ) { + return; + } + writeInstrumentationState((state) => { + const next = { ...state }; + for (const [key, record] of Object.entries(state)) { + if (key.startsWith(prefix) && record.attemptId === attemptId) delete next[key]; + } + return next; + }); +} + +export function releaseInstrumentationTurnState( + provider: string, + sessionId: string, + turnId?: string, +): void { + const prefix = `${provider}\0`; + const current = contextStorage.getStore()?.get(InstrumentationStateKey); + if (current === undefined) return; + const matches = (key: string, record: InstrumentationStateRecord): boolean => + key.startsWith(prefix) && + record.sessionId === sessionId && + (turnId === undefined || record.turnId === turnId); + if (!Object.entries(current).some(([key, record]) => matches(key, record))) return; + writeInstrumentationState((state) => { + const next = { ...state }; + for (const [key, record] of Object.entries(state)) { + if (matches(key, record)) delete next[key]; + } + return next; + }); +} + +/** Remembers where a durable runtime action originated. */ +export function rememberInstrumentationActionScope( + idempotencyKey: string, + scope: InstrumentationAttemptScope, +): void { + writeContextKey(InstrumentationActionScopeKey, (state) => ({ + ...state, + [idempotencyKey]: scope, + })); +} + +export interface InstrumentationActionCorrelation { + readonly idempotencyKey: string; + readonly scope: InstrumentationAttemptScope; +} + +export function findInstrumentationActionScopeForCall( + sessionId: string, + callId: string, +): InstrumentationActionCorrelation | undefined { + const scopes = contextStorage.getStore()?.get(InstrumentationActionScopeKey); + if (scopes === undefined) return undefined; + for (const scope of Object.values(scopes)) { + const idempotencyKey = `action:${sessionId}:${scope.turnId}:${callId}`; + if (scopes[idempotencyKey] !== undefined) return { idempotencyKey, scope }; + } + return undefined; +} + +/** Reads and releases one durable runtime action's originating scope. */ +export function takeInstrumentationActionScopeForCall( + sessionId: string, + callId: string, +): InstrumentationActionCorrelation | undefined { + const correlation = findInstrumentationActionScopeForCall(sessionId, callId); + if (correlation === undefined) return undefined; + writeContextKey(InstrumentationActionScopeKey, (state) => { + const next = { ...state }; + delete next[correlation.idempotencyKey]; + return next; + }); + return correlation; +} + +/** Takes every still-open action owned by one session or turn. */ +export function takeInstrumentationActionScopes( + sessionId: string, + turnId?: string, +): readonly InstrumentationActionCorrelation[] { + const current = contextStorage.getStore()?.get(InstrumentationActionScopeKey); + if (current === undefined) return []; + const correlations = Object.entries(current) + .filter( + ([, scope]) => + scope.sessionId === sessionId && (turnId === undefined || scope.turnId === turnId), + ) + .map(([idempotencyKey, scope]) => ({ idempotencyKey, scope })); + if (correlations.length === 0) return []; + const keys = new Set(correlations.map((correlation) => correlation.idempotencyKey)); + writeContextKey(InstrumentationActionScopeKey, (state) => { + const next = { ...state }; + for (const key of keys) delete next[key]; + return next; + }); + return correlations; +} + +function writeSlot( + state: InstrumentationStateMap, + key: string, + value: InstrumentationStateRecord | undefined, +): InstrumentationStateMap { + if (value === undefined) { + const next = { ...state }; + delete next[key]; + return next; + } + return { ...state, [key]: value }; +} + +function writeInstrumentationState( + update: (state: InstrumentationStateMap) => InstrumentationStateMap, +): void { + writeContextKey(InstrumentationStateKey, update); +} + +function assignOwner(record: InstrumentationStateRecord, owner: InstrumentationStateOwner): void { + if (owner.attemptId !== undefined) record.attemptId = owner.attemptId; + if (owner.sessionId !== undefined) record.sessionId = owner.sessionId; + if (owner.turnId !== undefined) record.turnId = owner.turnId; +} + +function writeContextKey>>( + key: ContextKey, + update: (state: T) => T, +): void { + if (contextStorage.getStore() === undefined) return; + loadContext().set(key, (state) => update(state ?? ({} as T))); +} + +/** A provider name cannot contain NUL, so the pair cannot be ambiguous. */ +function stateKey(provider: string, idempotencyKey: string): string { + return `${provider}\0${idempotencyKey}`; +} + +function deserializeState(data: unknown): InstrumentationStateMap { + if (typeof data !== "object" || data === null || Array.isArray(data)) return {}; + const state: Record = {}; + for (const [key, value] of Object.entries(data)) { + if (typeof value !== "object" || value === null || Array.isArray(value)) continue; + const record = value as Record; + const attemptId = typeof record["attemptId"] === "string" ? record["attemptId"] : undefined; + const sessionId = typeof record["sessionId"] === "string" ? record["sessionId"] : undefined; + const turnId = typeof record["turnId"] === "string" ? record["turnId"] : undefined; + const parsed: InstrumentationStateRecord = {}; + if (record["abandoned"] === true) parsed.abandoned = true; + if (attemptId !== undefined) parsed.attemptId = attemptId; + if (sessionId !== undefined) parsed.sessionId = sessionId; + if (turnId !== undefined) parsed.turnId = turnId; + if (record["value"] !== undefined) parsed.value = record["value"] as JsonValue; + state[key] = parsed; + } + return state; +} + +function deserializeActionScopes(data: unknown): InstrumentationActionScopeMap { + if (typeof data !== "object" || data === null || Array.isArray(data)) return {}; + return data as InstrumentationActionScopeMap; +} diff --git a/packages/eve/src/harness/runtime-actions.test.ts b/packages/eve/src/harness/runtime-actions.test.ts index 89241038f7..07afc1be0f 100644 --- a/packages/eve/src/harness/runtime-actions.test.ts +++ b/packages/eve/src/harness/runtime-actions.test.ts @@ -5,6 +5,7 @@ import { import { describe, expect, it } from "vitest"; import { + createRuntimeActionRequestFromToolCall, getPendingRuntimeActionBatch, resolvePendingRuntimeActions, resolveToolCallInputObject, @@ -35,6 +36,52 @@ const OPERATION_ID = deriveAgentOperationId({ parentTurnId: "turn_0", }); +describe("createRuntimeActionRequestFromToolCall", () => { + const loadSkillCall = { + input: { skill: "research" }, + toolCallId: "call-skill", + toolName: "load_skill", + type: "tool-call" as const, + }; + + it("classifies the framework load_skill tool as a skill action", () => { + expect( + createRuntimeActionRequestFromToolCall({ + toolCall: loadSkillCall, + tools: new Map([ + [ + "load_skill", + { + description: "Load a skill.", + frameworkAction: "load-skill" as const, + inputSchema: jsonSchema({ type: "object" }), + name: "load_skill", + }, + ], + ]), + }), + ).toEqual({ + callId: "call-skill", + input: { skill: "research" }, + kind: "load-skill", + }); + }); + + it("keeps an authored load_skill override as an ordinary tool action", () => { + expect( + createRuntimeActionRequestFromToolCall({ + toolCall: loadSkillCall, + tools: new Map(), + }), + ).toEqual({ + callId: "call-skill", + input: { skill: "research" }, + kind: "tool-call", + toolName: "load_skill", + }); + }); +}); + function createParkedSession(): HarnessSession { const base: HarnessSession = { agent: { modelReference: { id: "test-model" }, system: "", tools: [] }, @@ -611,3 +658,4 @@ describe("resolveToolCallInputObject", () => { ); }); }); +import { jsonSchema } from "ai"; diff --git a/packages/eve/src/harness/runtime-actions.ts b/packages/eve/src/harness/runtime-actions.ts index 05cdc445d5..f69f9c90a6 100644 --- a/packages/eve/src/harness/runtime-actions.ts +++ b/packages/eve/src/harness/runtime-actions.ts @@ -340,6 +340,17 @@ export function createRuntimeActionRequestFromToolCall(input: { }): RuntimeActionRequest { const definition = input.tools.get(input.toolCall.toolName); + if (definition?.frameworkAction === "load-skill") { + return { + callId: input.toolCall.toolCallId, + input: resolveToolCallInputObject(input.toolCall.input, { + callId: input.toolCall.toolCallId, + toolName: input.toolCall.toolName, + }), + kind: "load-skill", + }; + } + if (definition?.runtimeAction?.kind === "subagent-call") { return { callId: input.toolCall.toolCallId, diff --git a/packages/eve/src/harness/tool-loop.test.ts b/packages/eve/src/harness/tool-loop.test.ts index cc7190c511..c636a25051 100644 --- a/packages/eve/src/harness/tool-loop.test.ts +++ b/packages/eve/src/harness/tool-loop.test.ts @@ -9588,7 +9588,7 @@ describe("createToolLoopHarness", () => { }); const attemptCompleted = vi.fn(); const hooks = createInstrumentationHooks([ - { events: { "step.attempt.completed": attemptCompleted } }, + { events: { "step.attempt.completed": attemptCompleted }, name: "attempt" }, ]); const runInContext: InstrumentationContextRunner = (_operation, execute) => execute(); const config = createTestConfig("conversation", undefined, { @@ -9608,7 +9608,6 @@ describe("createToolLoopHarness", () => { }), hooks, runInContext, - expect.any(Function), ); const bridge = mockCreateAiSdkHookBridge.mock.results[0]!.value; const agentCall = vi.mocked(ToolLoopAgent).mock.calls[0]?.[0] as { @@ -9628,39 +9627,10 @@ describe("createToolLoopHarness", () => { scope: expect.objectContaining({ attemptIndex: 0 }), type: "step.attempt.completed", }), + expect.anything(), ); }); - it("resolves each action kind from the harness tool map", async () => { - setupMockAgent({ - finishReason: "stop", - response: { messages: [{ content: "Hello!", role: "assistant" }] }, - text: "Hello!", - toolCalls: [], - toolResults: [], - }); - const runStep = createToolLoopHarness( - createTestConfig("conversation", undefined, { - instrumentation: { - hooks: createInstrumentationHooks([]), - runInContext: (_operation, execute) => execute(), - }, - tools: createDelegationToolMap(), - }), - ); - - await runStep(createTestSession(), { message: "hi" }); - - const resolveActionKind = mockCreateAiSdkHookBridge.mock.calls[0]![3] as ( - toolName: string, - ) => string; - expect(resolveActionKind("delegate")).toBe("subagent-call"); - expect(resolveActionKind("add")).toBe("tool-call"); - // A name the harness never registered — a dynamic subagent resolved after - // the map was built lands here rather than throwing. - expect(resolveActionKind("absent")).toBe("tool-call"); - }); - it("publishes a delegation action when the AI SDK skips execution callbacks", async () => { setupMockAgent({ finishReason: "tool-calls", @@ -9691,7 +9661,9 @@ describe("createToolLoopHarness", () => { toolResults: [], }); const started = vi.fn(); - const hooks = createInstrumentationHooks([{ events: { "tool.call.started": started } }]); + const hooks = createInstrumentationHooks([ + { events: { "action.started": started }, name: "actions" }, + ]); const { emit } = createEventCollector(); const runStep = createToolLoopHarness( createTestConfig("conversation", emit, { @@ -9706,9 +9678,10 @@ describe("createToolLoopHarness", () => { expect.objectContaining({ callId: "call-delegate", kind: "subagent-call", - toolName: "delegate", - type: "tool.call.started", + name: "delegate", + type: "action.started", }), + expect.anything(), ); }); diff --git a/packages/eve/src/harness/tool-loop.ts b/packages/eve/src/harness/tool-loop.ts index f6c77d8205..23036f41df 100644 --- a/packages/eve/src/harness/tool-loop.ts +++ b/packages/eve/src/harness/tool-loop.ts @@ -113,12 +113,10 @@ import { import { createToolResultMessagePartFromToolError } from "#harness/action-result-helpers.js"; import { activeTurnId } from "#harness/active-turn-id.js"; import { buildTelemetryRuntimeContext } from "#harness/instrumentation-runtime-context.js"; -import { createAiSdkHookBridge, type ActionKindResolver } from "#harness/ai-sdk-hook-bridge.js"; -import { - createInstrumentationHandleEvent, - type InstrumentationActionSource, -} from "#harness/instrumentation-native-events.js"; +import { createAiSdkHookBridge } from "#harness/ai-sdk-hook-bridge.js"; +import { createInstrumentationHandleEvent } 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 { @@ -579,10 +577,10 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { let emissionState = getHarnessEmissionState(session.state); const store = contextStorage.getStore(); const parent = store?.get(ParentSessionKey); - let actionSource: InstrumentationActionSource | undefined; + let activeAttemptScope: InstrumentationAttemptScope | undefined; const emit = createInstrumentationHandleEvent({ agentName: config.runtimeIdentity?.agentName, - getActionSource: () => actionSource, + getAttemptScope: () => activeAttemptScope, handleEvent: baseEmit, hooks: config.instrumentation?.hooks, parentLineage: resolveParentLineage(parent, store?.get(ChannelKey)), @@ -1041,12 +1039,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { stepIndex: emissionState.stepIndex, turnId: instrumentationTurnId, }; - actionSource = - attemptScope === undefined - ? undefined - : { scope: attemptScope, tools: advertisedHarnessTools }; - const resolveActionKind: ActionKindResolver = (toolName) => - advertisedHarnessTools.get(toolName)?.runtimeAction?.kind ?? "tool-call"; + activeAttemptScope = attemptScope; const bridgeIntegration = attemptScope === undefined || instrumentationHooks === undefined ? undefined @@ -1054,7 +1047,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { attemptScope, instrumentationHooks, config.instrumentation?.runInContext, - resolveActionKind, ); const hooks = buildStepHooks({ @@ -1184,6 +1176,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 +1186,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/internal/application/compiled-artifacts.ts b/packages/eve/src/internal/application/compiled-artifacts.ts index ba835e0e19..ad1487eece 100644 --- a/packages/eve/src/internal/application/compiled-artifacts.ts +++ b/packages/eve/src/internal/application/compiled-artifacts.ts @@ -480,6 +480,7 @@ function createInstrumentationPluginSource(input: { "// Default export satisfies the Nitro plugin contract so this file", "// can be used directly as a Nitro plugin without a separate wrapper.", "export default function installInstrumentationPlugin(nitroApp) {", + " // The last point a buffered exporter can still reach the network.", " nitroApp?.hooks?.hook('close', async () => {", " await shutdownInstrumentationProviders();", " });", diff --git a/packages/eve/src/internal/application/dev-environment.ts b/packages/eve/src/internal/application/dev-environment.ts index acbd09cfce..daa2ff244f 100644 --- a/packages/eve/src/internal/application/dev-environment.ts +++ b/packages/eve/src/internal/application/dev-environment.ts @@ -5,3 +5,25 @@ export const EVE_DEV_ENV_FLAG = "EVE_DEV"; export function isEveDevEnvironment(): boolean { return process.env[EVE_DEV_ENV_FLAG] === "1"; } + +/** Environment flag set for a server `eve eval` started to run against. */ +export const EVE_EVALUATION_ENV_FLAG = "EVE_EVALUATION"; + +/** Stable identifier for the local eval run this server was started to serve. */ +export const EVE_EVALUATION_RUN_ID_ENV = "EVE_EVALUATION_RUN_ID"; + +/** + * Reports whether this process exists to serve an eval run. + * + * False for a server that `eve eval --url` merely points at: that process was + * started to serve ordinary traffic and cannot know an eval is among it. + */ +export function isEveEvaluationEnvironment(): boolean { + return process.env[EVE_EVALUATION_ENV_FLAG] === "1"; +} + +export function resolveEveEvaluationRunId(): string | undefined { + if (!isEveEvaluationEnvironment()) return undefined; + const runId = process.env[EVE_EVALUATION_RUN_ID_ENV]; + return runId === undefined || runId.length === 0 ? undefined : runId; +} diff --git a/packages/eve/src/public/instrumentation/provider.ts b/packages/eve/src/public/instrumentation/provider.ts index fd603f0981..e4fa4033d4 100644 --- a/packages/eve/src/public/instrumentation/provider.ts +++ b/packages/eve/src/public/instrumentation/provider.ts @@ -10,9 +10,16 @@ // from the union below is what keeps the public contract from drifting away // from the bus that feeds it. import type { InstrumentationEvent } from "#harness/instrumentation-lifecycle.js"; +import type { JsonValue } from "#public/types/json.js"; + +export type { JsonValue } from "#public/types/json.js"; export type { + InstrumentationActionCompletedEvent, + InstrumentationActionFailedEvent, InstrumentationActionKind, + InstrumentationActionOutput, + InstrumentationActionStartedEvent, InstrumentationAttemptScope, InstrumentationContentPart, InstrumentationEvent, @@ -22,16 +29,23 @@ export type { InstrumentationModelRef, InstrumentationOperationRef, InstrumentationParentLineage, + InstrumentationSessionFailedEvent, + InstrumentationSessionSettledEvent, InstrumentationSessionStartedEvent, InstrumentationSessionTransitionEvent, + InstrumentationStepAttemptCompletedEvent, + InstrumentationStepAttemptFailedEvent, InstrumentationStepAttemptMetadataEvent, InstrumentationStepAttemptStartedEvent, InstrumentationStepAttemptTerminalEvent, + InstrumentationTraceContext, InstrumentationToolCallCompletedEvent, InstrumentationToolCallFailedEvent, InstrumentationToolCallStartedEvent, + InstrumentationToolCallTerminalEvent, InstrumentationToolOutput, - InstrumentationTraceContext, + InstrumentationTurnFailedEvent, + InstrumentationTurnSettledEvent, InstrumentationTurnStartedEvent, InstrumentationTurnTerminalEvent, InstrumentationUsage, @@ -56,6 +70,11 @@ export const DISABLED = Symbol.for("eve.instrumentation.disabled"); /** Where the agent is running when `setup` fires. */ export type InstrumentationEnvironment = "development" | "preview" | "production"; +/** The local eval run this server was started to serve. */ +export interface EvaluationRef { + readonly runId: string; +} + /** * Passed to {@link InstrumentationProvider.setup} once at server startup, * before any event is published. @@ -64,24 +83,63 @@ export interface ProviderSetupContext { /** The agent name declared by `defineAgent`. */ readonly agentName: string; readonly environment: InstrumentationEnvironment; + /** + * The `eve eval` run this server exists to serve. + * + * Eval traffic is synthetic, so a provider billed per span or feeding a + * production dashboard usually wants to return early rather than export it. + * + * Absent for ordinary servers and for a server that `eve eval --url` merely + * points at. That process cannot claim one remote caller's run as its own. + */ + readonly evaluation?: EvaluationRef; /** The eve version running the agent. */ readonly frameworkVersion: string; } /** - * The second argument to every handler. + * Durable state for one provider and one operation. + * + * Scoped to both, so two providers reacting to the same event cannot see or + * overwrite each other, and one provider's turns and actions stay separate. * - * Empty today. It exists now so adding durable per-provider state later is an - * additive change rather than a second break in every handler signature. + * It survives a step boundary. An approval-gated tool suspends and resumes in a + * different process; a value written at `action.started` is still readable at + * `action.completed` there, which a plain module-level `Map` would not be. */ -export type ProviderContext = Readonly>; +export interface ProviderState { + get(): JsonValue | undefined; + /** + * Stages a value; `undefined` releases the slot. + * + * Synchronous because it writes into the durable context eve commits when the + * step settles, not to a store of its own. A value that cannot survive JSON + * throws here rather than at the step boundary. Writes after this handler + * settles or times out are ignored. + */ + set(value: JsonValue | undefined): void; +} + +/** The second argument to every handler. */ +export interface ProviderContext { + readonly state: ProviderState; +} /** * One event handler. * - * eve balances every start with exactly one terminal, so a handler that needs - * to carry a value from a start to its terminal can key its own map on - * `event.id` and delete on the terminal. + * A handler that needs to carry a durable value from a start to its terminal + * writes it to `ctx.state` and reads it back on the terminal. eve releases the + * slot when that terminal arrives. + * + * The AI SDK can omit a model terminal when an incomplete stream closes. eve + * releases any remaining model state when the step attempt terminates. + * + * A handler that does not settle within eve's timeout is abandoned: it keeps + * running, but the rest of that operation is withheld from the provider, + * including the terminal. eve releases the state slot in that case too, so an + * abandoned handler leaks nothing — but a provider holding a resource of its + * own outside `ctx.state` still needs its own expiry. */ export type Handler = (event: TEvent, ctx: ProviderContext) => void | PromiseLike; 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 04d2bdfd21..6ccbc96bd6 100644 --- a/packages/eve/src/tracing/agent-otel-provider.test.ts +++ b/packages/eve/src/tracing/agent-otel-provider.test.ts @@ -7,21 +7,32 @@ import { } from "@opentelemetry/sdk-trace-base"; import { describe, expect, it } from "vitest"; -import { createAiSdkHookBridge, type ActionKindResolver } from "#harness/ai-sdk-hook-bridge.js"; +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 { createInstrumentationHooks, + 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; @@ -30,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({ @@ -52,7 +65,7 @@ async function emitAttempt(input: { readonly hooks: InstrumentationHooks; readonly runInContext: InstrumentationContextRunner; readonly providerMetadata?: Readonly>; - readonly resolveActionKind?: ActionKindResolver; + readonly actionKind?: InstrumentationActionKind; readonly sessionId: string; readonly skipModelTerminal?: boolean; readonly skipToolTerminal?: boolean; @@ -73,12 +86,7 @@ async function emitAttempt(input: { await publishTurnStarted(input); } - const bridge = createAiSdkHookBridge( - scope, - input.hooks, - input.runInContext, - input.resolveActionKind, - ); + const bridge = createAiSdkHookBridge(scope, input.hooks, input.runInContext); Reflect.apply(bridge.onStart!, bridge, [ { callId: "call-1", @@ -140,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", @@ -164,23 +182,39 @@ 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) { 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 +234,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 +258,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[] { @@ -296,7 +342,100 @@ describe("createAgentOtelInstrumentation", () => { }); }); - 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, @@ -311,7 +450,7 @@ describe("createAgentOtelInstrumentation", () => { const spans = runtime.exporter.getFinishedSpans(); expect(byName(spans, "ai.streamText.doStream")).toHaveLength(1); - expect(byName(spans, "agent.action")).toHaveLength(1); + expect(byName(spans, "agent.action")).toHaveLength(0); expect(byName(spans, "ai.toolCall")).toHaveLength(1); expect(byName(spans, "agent.step")).toHaveLength(1); }); @@ -320,7 +459,7 @@ describe("createAgentOtelInstrumentation", () => { const runtime = createRuntime(); await emitAttempt({ hooks: runtime.hooks, - resolveActionKind: () => "subagent-call", + actionKind: "subagent-call", runInContext: runtime.runInContext, sessionId: "session-1", turnId: "turn-1", @@ -369,10 +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}'); - // Structural spans stay structural: content lives only on the operation spans. - const structural = byName(spans, "agent.action")[0]!; - expect(JSON.stringify(structural.attributes)).not.toContain("secret"); - expect(JSON.stringify(structural.attributes)).not.toContain("temperature"); + // 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 () => { @@ -392,11 +531,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..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, @@ -41,6 +42,7 @@ import type { InstrumentationTurnTerminalEvent, InstrumentationUsage, } from "#harness/instrumentation-lifecycle.js"; +import { sessionIdempotencyKey } from "#harness/instrumentation-lifecycle.js"; interface SpanState { readonly context: Context; @@ -52,9 +54,7 @@ interface AttemptSpanState { readonly step: SpanState; } -interface ToolSpanState extends SpanState { - readonly toolSpan: Span; -} +type ToolSpanState = SpanState; export interface AgentOtelInstrumentationInput { /** @@ -93,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); @@ -104,6 +118,7 @@ export function createAgentOtelInstrumentation( await ensureSessionContext({ agentName: undefined, channelKind: undefined, + idempotencyKey: sessionIdempotencyKey(event.sessionId), parentTraceContext: event.parentTraceContext, rootSessionId: event.rootSessionId, sessionId: event.sessionId, @@ -197,11 +212,17 @@ 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, { ...turn, - terminal: { error: event.error, type: event.type }, + terminal: + event.type === "turn.failed" + ? { error: event.error, type: event.type } + : { type: event.type }, }); }; @@ -253,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); } }; @@ -280,13 +302,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); @@ -337,29 +359,13 @@ 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 actionSpan = input.tracer.startSpan( - "agent.action", - { - attributes: { - "agent.action.call_id": event.callId, - "agent.action.kind": event.kind, - "agent.action.name": event.toolName, - "agent.framework.name": "eve", - "agent.framework.version": input.frameworkVersion, - "agent.root.session.id": event.scope.rootSessionId ?? event.scope.sessionId, - "agent.session.id": event.scope.sessionId, - "agent.step.attempt": event.scope.attemptIndex, - "agent.step.index": event.scope.stepIndex, - "agent.turn.id": event.scope.turnId, - }, - }, - attempt.step.context, - ); - const actionContext = trace.setSpan(attempt.step.context, actionSpan); - const toolSpan = input.tracer.startSpan( + 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", { attributes: { @@ -368,36 +374,29 @@ export function createAgentOtelInstrumentation( "gen_ai.tool.name": event.toolName, }, }, - actionContext, + parentContext, ); if (recordInputs) { const args = contentAttribute(event.input, false); - if (args !== undefined) toolSpan.setAttribute("gen_ai.tool.call.arguments", args); + if (args !== undefined) span.setAttribute("gen_ai.tool.call.arguments", args); } - const state: ToolSpanState = { - context: trace.setSpan(actionContext, toolSpan), - span: actionSpan, - toolSpan, - }; - getExecutionContexts(event.scope).tools.set(event.id, state.context); - getSpanStates(toolSpans, event.scope).set(event.id, state); + 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); }; 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); recordError(state.span, event.error); } else if (event.output.type === "error") { - recordError(state.toolSpan, event.output.error); recordError(state.span, event.output.error); } else if (recordOutputs) { const result = contentAttribute(event.output.output, false); - if (result !== undefined) state.toolSpan.setAttribute("gen_ai.tool.call.result", result); + if (result !== undefined) state.span.setAttribute("gen_ai.tool.call.result", result); } - state.toolSpan.end(); state.span.end(); }; @@ -492,6 +491,7 @@ export function createAgentOtelInstrumentation( return { hook: { events: { + ...actions.events, "step.attempt.completed": onStepTerminal, "step.attempt.failed": onStepTerminal, "step.attempt.metadata": onStepMetadata, @@ -511,13 +511,14 @@ export function createAgentOtelInstrumentation( "turn.failed": onTurnTerminal, "turn.started": onTurnStarted, }, + name: "eve.otel", }, runInContext(operation, execute) { 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); }, }; @@ -537,10 +538,7 @@ export function createAgentOtelInstrumentation( function drainOpenSpans(scope: InstrumentationAttemptScope): void { for (const state of modelSpans.get(scope)?.values() ?? []) state.span.end(); modelSpans.delete(scope); - for (const state of toolSpans.get(scope)?.values() ?? []) { - state.toolSpan.end(); - state.span.end(); - } + for (const state of toolSpans.get(scope)?.values() ?? []) state.span.end(); toolSpans.delete(scope); } } 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..470a35a821 100644 --- a/packages/eve/src/tracing/agent-trace-context-store.test.ts +++ b/packages/eve/src/tracing/agent-trace-context-store.test.ts @@ -52,9 +52,7 @@ describe("ContextAgentTraceStateStore", () => { }, parentSpanId: "2".repeat(16), startTimeMs: 1_700_000_000_000, - }); - expect(store.getTurn("session-1", "turn-1")?.terminal?.error).toMatchObject({ - message: "failed", + terminal: { error: { message: "failed" }, type: "turn.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..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, @@ -112,10 +154,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 }, }, ]), ), @@ -123,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 { @@ -150,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( @@ -187,7 +271,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..49c7c5cbf9 100644 --- a/packages/eve/src/tracing/agent-trace-state.ts +++ b/packages/eve/src/tracing/agent-trace-state.ts @@ -1,8 +1,11 @@ import type { SpanContext } from "#compiled/@opentelemetry/api/index.js"; import type { + InstrumentationActionKind, InstrumentationParentLineage, - InstrumentationTurnTerminalEvent, + InstrumentationTraceContext, + InstrumentationTurnFailedEvent, + InstrumentationTurnSettledEvent, } from "#harness/instrumentation-lifecycle.js"; /** Sized so an ordinary session stays one trace and only an outsized one rolls. */ @@ -24,16 +27,39 @@ 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"] }; +} + +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/install-instrumentation-runtime.test.ts b/packages/eve/src/tracing/install-instrumentation-runtime.test.ts index 2c3fef8a8b..6c0b35bd99 100644 --- a/packages/eve/src/tracing/install-instrumentation-runtime.test.ts +++ b/packages/eve/src/tracing/install-instrumentation-runtime.test.ts @@ -45,7 +45,7 @@ describe("installInstrumentationRuntime", () => { const runtime = installInstrumentationRuntime({ collected: collectOtelPipeline([otelIntegration()]), frameworkVersion: "test", - providers: [{ flush: providerFlush, shutdown: providerShutdown }], + providers: [{ flush: providerFlush, name: "test", shutdown: providerShutdown }], serviceName: "weather", }); diff --git a/packages/eve/src/tracing/install-instrumentation-runtime.ts b/packages/eve/src/tracing/install-instrumentation-runtime.ts index 9abc32eafb..e334b864e1 100644 --- a/packages/eve/src/tracing/install-instrumentation-runtime.ts +++ b/packages/eve/src/tracing/install-instrumentation-runtime.ts @@ -19,7 +19,16 @@ import { registerOtelPipeline, type RegisteredOtelPipeline } from "#tracing/otel const log = createLogger("tracing.install-instrumentation-runtime"); -/** Installs the bus and the one OpenTelemetry pipeline collected for this process. */ +/** + * Installs the process instrumentation runtime around a collected pipeline. + * + * Both layouts land here. `eve dev`'s zero-config default and an authored + * `agent/instrumentation/` directory differ only in where the declared values + * came from, so sharing the install keeps them on one runtime path. + * + * A directory that declared no OpenTelemetry still gets a bus: its providers + * see every event, they just have no spans to hang them on. + */ export function installInstrumentationRuntime(input: { readonly collected: CollectedOtel; readonly frameworkVersion: string; @@ -44,6 +53,7 @@ export function installInstrumentationRuntime(input: { stateStore: new ContextAgentTraceStateStore(), tracer: trace.getTracer("eve.agent", input.frameworkVersion), }); + // First, so the span is open before an authored provider sees the event. providers.unshift(agentOtel.hook); runInContext = agentOtel.runInContext; 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..99185b6bff 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,12 @@ 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 { + actionIdempotencyKey, + 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 +49,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", @@ -85,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", @@ -109,14 +127,26 @@ 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: actionKey, + output: { output: { temperature: 72 }, type: "result" }, + scope, + type: "action.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", diff --git a/packages/eve/src/tracing/otel-registration.scenario.test.ts b/packages/eve/src/tracing/otel-registration.scenario.test.ts index 204b34c81c..530222a325 100644 --- a/packages/eve/src/tracing/otel-registration.scenario.test.ts +++ b/packages/eve/src/tracing/otel-registration.scenario.test.ts @@ -1,6 +1,6 @@ import { context, propagation, trace, type Context } from "@opentelemetry/api"; import { InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { registerOtelPipeline } from "#tracing/otel-registration.js"; @@ -40,7 +40,8 @@ describe("registerOtelPipeline", () => { it("does not export the private registration span", async () => { const exporter = new InMemorySpanExporter(); const processor = new SimpleSpanProcessor(exporter); - registerOtelPipeline({ + const shutdown = vi.spyOn(processor, "shutdown"); + const runtime = registerOtelPipeline({ pipeline: { spanProcessors: [processor] }, serviceName: "weather", }); @@ -49,6 +50,7 @@ describe("registerOtelPipeline", () => { await processor.forceFlush(); expect(exporter.getFinishedSpans().map((span) => span.name)).toEqual(["user.work"]); - await processor.shutdown(); + await runtime.shutdown(); + expect(shutdown).toHaveBeenCalledOnce(); }); }); diff --git a/packages/eve/test/scenarios/eval-command-environment.scenario.test.ts b/packages/eve/test/scenarios/eval-command-environment.scenario.test.ts index 875a3c4196..d6dcd70c05 100644 --- a/packages/eve/test/scenarios/eval-command-environment.scenario.test.ts +++ b/packages/eve/test/scenarios/eval-command-environment.scenario.test.ts @@ -44,6 +44,8 @@ const DEVELOPMENT_ENV_KEYS = [ "EVE_DEV_LOCAL_ONLY", "EVE_DEV_SHARED", "EVE_DEV_SHELL_ONLY", + "EVE_EVALUATION", + "EVE_EVALUATION_RUN_ID", ] as const; async function createEnvironmentFixture(): Promise { @@ -192,6 +194,8 @@ describe("eve eval environment loading", () => { expect(close).toHaveBeenCalledTimes(1); expect(handle.shutdown).toHaveBeenCalledTimes(1); + expect(process.env.EVE_EVALUATION).toBe("1"); + expect(process.env.EVE_EVALUATION_RUN_ID).toMatch(/^[0-9a-f-]{36}$/u); expect(exit).toHaveBeenCalledWith(0); });