From 439463d0a228255c7acb3674b8072f219134e670 Mon Sep 17 00:00:00 2001 From: Chad Hietala Date: Sat, 8 Aug 2026 17:17:53 -0400 Subject: [PATCH] feat(eve): publish durable runtime action events Signed-off-by: Chad Hietala --- .changeset/instrumentation-runtime-actions.md | 5 + .../src/evals/runner/derive-run-facts.test.ts | 33 +++ .../eve/src/evals/runner/derive-run-facts.ts | 8 +- packages/eve/src/execution/node-step.ts | 3 + .../src/harness/ai-sdk-hook-bridge.test.ts | 20 -- .../eve/src/harness/ai-sdk-hook-bridge.ts | 13 - packages/eve/src/harness/execute-tool.ts | 1 + .../harness/instrumentation-lifecycle.test.ts | 69 ++++- .../src/harness/instrumentation-lifecycle.ts | 181 +++++++++++-- .../instrumentation-native-events.test.ts | 159 ++++++----- .../harness/instrumentation-native-events.ts | 95 +++++-- .../src/harness/instrumentation-state.test.ts | 10 +- .../eve/src/harness/instrumentation-state.ts | 252 +++++++++++++++--- .../eve/src/harness/runtime-actions.test.ts | 48 ++++ packages/eve/src/harness/runtime-actions.ts | 11 + packages/eve/src/harness/tool-loop.test.ts | 37 +-- packages/eve/src/harness/tool-loop.ts | 19 +- .../src/public/instrumentation/provider.ts | 4 + .../src/tracing/agent-otel-provider.test.ts | 51 +--- .../eve/src/tracing/agent-otel-provider.ts | 44 +-- ...l-instrumentation-runtime.scenario.test.ts | 4 +- 21 files changed, 753 insertions(+), 314 deletions(-) create mode 100644 .changeset/instrumentation-runtime-actions.md diff --git a/.changeset/instrumentation-runtime-actions.md b/.changeset/instrumentation-runtime-actions.md new file mode 100644 index 0000000000..5123018264 --- /dev/null +++ b/.changeset/instrumentation-runtime-actions.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Publish durable action lifecycle events for tools, skills, subagents, and remote agents while preserving the AI SDK tool-call boundary separately. diff --git a/packages/eve/src/evals/runner/derive-run-facts.test.ts b/packages/eve/src/evals/runner/derive-run-facts.test.ts index 6bb14533ff..b45803288a 100644 --- a/packages/eve/src/evals/runner/derive-run-facts.test.ts +++ b/packages/eve/src/evals/runner/derive-run-facts.test.ts @@ -162,6 +162,39 @@ describe("deriveRunFacts", () => { expect(facts.toolCallCount).toBe(2); }); + it("preserves framework skill input in the eval tool-call view", () => { + const facts = derive([ + turnStarted("t1", 0), + { + type: "actions.requested", + data: { + actions: [ + { + callId: "skill-1", + input: { skill: "research" }, + kind: "load-skill", + }, + ], + sequence: 1, + stepIndex: 0, + turnId: "t1", + }, + }, + actionResult({ callId: "skill-1", toolName: "load_skill", output: "Skill body" }), + ]); + + expect(facts.toolCalls).toEqual([ + { + input: { skill: "research" }, + name: "load_skill", + output: "Skill body", + sessionId: undefined, + status: "completed", + turnIndex: 0, + }, + ]); + }); + it("uses the normalized failed lifecycle status for error results", () => { const events: UnstampedMessageStreamEvent[] = [ actionsRequested([{ callId: "c1", toolName: "bash" }]), diff --git a/packages/eve/src/evals/runner/derive-run-facts.ts b/packages/eve/src/evals/runner/derive-run-facts.ts index 5d6378a338..20578a54d9 100644 --- a/packages/eve/src/evals/runner/derive-run-facts.ts +++ b/packages/eve/src/evals/runner/derive-run-facts.ts @@ -1,4 +1,5 @@ import type { MessageStreamEvent } from "#protocol/message.js"; +import { LOAD_SKILL_TOOL_NAME } from "#runtime/skills/fragment-context.js"; import type { InputRequest } from "#runtime/input/types.js"; import type { JsonObject, JsonValue } from "#shared/json.js"; import type { EveEvalDerivedFacts, EveEvalSubagentCall, EveEvalToolCall } from "#evals/types.js"; @@ -107,8 +108,11 @@ export function deriveRunFacts( case "actions.requested": { for (const action of event.data.actions) { - if (action.kind !== "tool-call") continue; - ensureToolCall(action.callId, action.toolName, action.input); + if (action.kind === "tool-call") { + ensureToolCall(action.callId, action.toolName, action.input); + } else if (action.kind === "load-skill") { + ensureToolCall(action.callId, LOAD_SKILL_TOOL_NAME, action.input); + } } break; } 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/harness/ai-sdk-hook-bridge.test.ts b/packages/eve/src/harness/ai-sdk-hook-bridge.test.ts index b8dfed0b52..7654319cb1 100644 --- a/packages/eve/src/harness/ai-sdk-hook-bridge.test.ts +++ b/packages/eve/src/harness/ai-sdk-hook-bridge.test.ts @@ -379,7 +379,6 @@ describe("createAiSdkHookBridge", () => { callId: "tool-1", idempotencyKey: `tool:${scope.attemptId}:tool-1:0`, input: { q: "eve" }, - kind: "tool-call", scope, toolName: "search", type: "tool.call.started", @@ -398,25 +397,6 @@ describe("createAiSdkHookBridge", () => { }, ); - 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 => { diff --git a/packages/eve/src/harness/ai-sdk-hook-bridge.ts b/packages/eve/src/harness/ai-sdk-hook-bridge.ts index 9ebee70814..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, @@ -23,15 +22,8 @@ import { 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 modelKeys: Map; - readonly resolveActionKind: ActionKindResolver; readonly scope: InstrumentationAttemptScope; readonly toolKeys: Map; operation?: InstrumentationOperationRef; @@ -44,11 +36,9 @@ export function createAiSdkHookBridge( scope: InstrumentationAttemptScope, hooks: InstrumentationHooks, runInContext: InstrumentationContextRunner = directRunInContext, - resolveActionKind: ActionKindResolver = defaultResolveActionKind, ): Telemetry { const state: AttemptState = { modelKeys: new Map(), - resolveActionKind, scope, toolKeys: new Map(), }; @@ -151,8 +141,6 @@ export function createAiSdkHookBridge( const directRunInContext: InstrumentationContextRunner = (_operation, execute) => execute(); -const defaultResolveActionKind: ActionKindResolver = () => "tool-call"; - function toStepAttemptStarted( state: AttemptState, ): InstrumentationStepAttemptStartedEvent | undefined { @@ -260,7 +248,6 @@ function toToolCallStarted( callId: source.toolCall.toolCallId, idempotencyKey, input: source.toolCall.input, - kind: state.resolveActionKind(source.toolCall.toolName), scope: state.scope, toolName: source.toolCall.toolName, type: "tool.call.started", 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 index f08e2de5a9..c070e6226f 100644 --- a/packages/eve/src/harness/instrumentation-lifecycle.test.ts +++ b/packages/eve/src/harness/instrumentation-lifecycle.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest"; import { ContextContainer, contextStorage } from "#context/container.js"; import { deserializeContext, serializeContext } from "#context/serialize.js"; import { + actionIdempotencyKey, attemptIdempotencyKey, createInstrumentationHooks, modelCallIdempotencyKey, @@ -11,7 +12,11 @@ import { turnIdempotencyKey, type InstrumentationAttemptScope, } from "#harness/instrumentation-lifecycle.js"; -import { instrumentationStateSlot } from "#harness/instrumentation-state.js"; +import { + findInstrumentationActionScopeForCall, + instrumentationStateSlot, + rememberInstrumentationActionScope, +} from "#harness/instrumentation-state.js"; const scope: InstrumentationAttemptScope = { attemptId: "session-1:turn-1:0:0", @@ -116,6 +121,68 @@ describe("provider state lifecycle", () => { expect(instrumentationStateSlot("sink", modelKey).get()).toBeUndefined(); }); }); + + it("keeps action state past 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("terminalizes and releases pending actions when a 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: turnIdempotencyKey(scope.sessionId, 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).toHaveBeenCalledOnce(); + }); }); describe("provider handler deadlines", () => { diff --git a/packages/eve/src/harness/instrumentation-lifecycle.ts b/packages/eve/src/harness/instrumentation-lifecycle.ts index 9fe97b883a..072b9e895b 100644 --- a/packages/eve/src/harness/instrumentation-lifecycle.ts +++ b/packages/eve/src/harness/instrumentation-lifecycle.ts @@ -1,12 +1,15 @@ -import { createLogger, formatError } from "#internal/logging.js"; import { abandonInstrumentationState, instrumentationStateSlot, isInstrumentationStateAbandoned, releaseInstrumentationAttemptState, - releaseInstrumentationState, + releaseInstrumentationTurnState, + takeInstrumentationActionScopes, + type InstrumentationStateOwner, type InstrumentationStateSlot, + releaseInstrumentationState, } from "#harness/instrumentation-state.js"; +import { createLogger, formatError } from "#internal/logging.js"; /** * Stable eve identity for one actual model attempt. @@ -78,18 +81,30 @@ 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 }; -/** Replay-stable row identity for every lifecycle operation. */ +/** + * 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}`; } @@ -118,6 +133,11 @@ export function toolCallIdempotencyKey( 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; @@ -189,6 +209,13 @@ export interface InstrumentationTurnStartedEvent { readonly turnId: string; } +/** + * 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; @@ -265,12 +292,13 @@ export type InstrumentationModelCallTerminalEvent = | InstrumentationModelCallCompletedEvent | InstrumentationModelCallFailedEvent; +export type InstrumentationToolOutput = InstrumentationActionOutput; + export interface InstrumentationToolCallStartedEvent { readonly type: "tool.call.started"; readonly callId: string; readonly idempotencyKey: string; readonly input: unknown; - readonly kind: InstrumentationActionKind; readonly scope: InstrumentationAttemptScope; readonly toolName: string; } @@ -293,14 +321,50 @@ 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, @@ -322,6 +386,9 @@ 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; @@ -330,7 +397,9 @@ export interface InstrumentationProviderDefinition { readonly "turn.failed"?: InstrumentationEventHandler; readonly "turn.started"?: InstrumentationEventHandler; }; + /** Drains anything buffered. Driven by the runtime, not by the bus. */ readonly flush?: () => void | PromiseLike; + /** Releases resources when the process is going away. */ readonly shutdown?: () => void | PromiseLike; } @@ -340,6 +409,8 @@ type InstrumentationProviderInput = Omit => { const terminal = isTerminal(event.type); const startedBoundary = event.type.endsWith(".started"); const attemptTerminal = event.type === "step.attempt.completed" || event.type === "step.attempt.failed"; - const attemptId = stateAttemptId(event); + 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 [providerIndex, provider] of providers.entries()) { const providerName = provider.name ?? `provider-${String(providerIndex)}`; + // 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(providerName, event.idempotencyKey); if (attemptTerminal) releaseInstrumentationAttemptState(providerName, event.scope.attemptId); + if (cleanupSession) releaseInstrumentationTurnState(providerName, event.sessionId); + if (cleanupTurn) + releaseInstrumentationTurnState(providerName, event.sessionId, event.turnId); }; + if (isInstrumentationStateAbandoned(providerName, event.idempotencyKey)) { release(); continue; } + const handler = provider.events?.[event.type]; if (handler === undefined) { release(); continue; } - const state = instrumentationStateSlot(providerName, event.idempotencyKey, attemptId); + + const state = instrumentationStateSlot(providerName, event.idempotencyKey, owner); + const ctx: InstrumentationHandlerContext = { state }; + try { const settled = await withTimeout( - () => (handler as InstrumentationEventHandler)(event, { state }), + () => (handler as InstrumentationEventHandler)(event, ctx), handlerTimeoutMs, () => { state.revoke(); if (startedBoundary) { - abandonInstrumentationState(providerName, event.idempotencyKey, attemptId); + abandonInstrumentationState(providerName, 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, @@ -451,6 +561,7 @@ export function createInstrumentationHooks( return { publish }; } +/** Resolves false when the deadline wins; rejects with whatever the handler threw. */ async function withTimeout( run: () => void | PromiseLike, timeoutMs: number, @@ -472,15 +583,35 @@ async function withTimeout( } } -function stateAttemptId(event: InstrumentationEvent): string | undefined { - if (!("scope" in event)) return undefined; +/** 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.") - ? event.scope.attemptId - : undefined; -} - + ? { 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 7f87122a75..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 () => { @@ -86,7 +93,7 @@ describe("createInstrumentationHandleEvent", () => { expect(events).toEqual([ { - idempotencyKey: "session:session-1", + idempotencyKey: sessionIdempotencyKey("session-1"), sessionId: "session-1", turnId: "turn-1", type: "session.waiting", @@ -119,7 +126,7 @@ describe("createInstrumentationHandleEvent", () => { expect(events.filter((event) => event.type === "turn.started")).toEqual([ { - idempotencyKey: "turn:child-1:child-turn-1", + idempotencyKey: turnIdempotencyKey("child-1", "child-turn-1"), parentLineage, parentTraceContext: undefined, rootSessionId: "session-1", @@ -129,7 +136,7 @@ describe("createInstrumentationHandleEvent", () => { type: "turn.started", }, { - idempotencyKey: "turn:child-1:child-turn-2", + idempotencyKey: turnIdempotencyKey("child-1", "child-turn-2"), parentLineage, parentTraceContext: undefined, rootSessionId: "session-1", @@ -141,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", @@ -150,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: [ { @@ -205,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.", @@ -221,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", - idempotencyKey: "tool:session-1:turn-1:0:0:delegate-1:0", + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "delegate-1"), input: { task: "research" }, kind: "subagent-call", + name: "delegate", scope, - toolName: "delegate", - type: "tool.call.started", + type: "action.started", + }, + { + callId: "skill-1", + idempotencyKey: actionIdempotencyKey("session-1", "turn-1", "skill-1"), + input: { name: "research" }, + kind: "load-skill", + name: "load_skill", + scope, + type: "action.started", }, { callId: "remote-1", - idempotencyKey: "tool:session-1:turn-1:0:0: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 98161c3c6c..63a06ec480 100644 --- a/packages/eve/src/harness/instrumentation-native-events.ts +++ b/packages/eve/src/harness/instrumentation-native-events.ts @@ -1,27 +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 { + actionIdempotencyKey, sessionIdempotencyKey, - toolCallIdempotencyKey, turnIdempotencyKey, } from "#harness/instrumentation-lifecycle.js"; -import type { HandleEventFn, HarnessToolMap } from "#harness/types.js"; - -export interface InstrumentationActionSource { - readonly scope: InstrumentationAttemptScope; - readonly tools: HarnessToolMap; -} +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; @@ -48,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, - idempotencyKey: toolCallIdempotencyKey(source.scope, 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( diff --git a/packages/eve/src/harness/instrumentation-state.test.ts b/packages/eve/src/harness/instrumentation-state.test.ts index 32f80eb005..7f7f4489ea 100644 --- a/packages/eve/src/harness/instrumentation-state.test.ts +++ b/packages/eve/src/harness/instrumentation-state.test.ts @@ -15,7 +15,9 @@ describe("instrumentation state", () => { it("survives a serialized step boundary", async () => { const context = new ContextContainer(); contextStorage.run(context, () => { - instrumentationStateSlot("sink", "model:1", "attempt-1").set({ rowId: "row-1" }); + instrumentationStateSlot("sink", "model:1", { attemptId: "attempt-1" }).set({ + rowId: "row-1", + }); }); const restored = await deserializeContext(await serializeContext(context)); contextStorage.run(restored, () => { @@ -55,8 +57,8 @@ describe("instrumentation state", () => { it("releases exact and attempt-owned state", () => { contextStorage.run(new ContextContainer(), () => { - instrumentationStateSlot("sink", "model:1", "attempt-1").set("one"); - instrumentationStateSlot("sink", "model:2", "attempt-2").set("two"); + instrumentationStateSlot("sink", "model:1", { attemptId: "attempt-1" }).set("one"); + instrumentationStateSlot("sink", "model:2", { attemptId: "attempt-2" }).set("two"); releaseInstrumentationState("sink", "model:2"); releaseInstrumentationAttemptState("sink", "attempt-1"); expect(instrumentationStateSlot("sink", "model:1").get()).toBeUndefined(); @@ -79,7 +81,7 @@ describe("instrumentation state", () => { it("persists abandonment across serialization", async () => { const context = new ContextContainer(); contextStorage.run(context, () => { - abandonInstrumentationState("sink", "model:1", "attempt-1"); + abandonInstrumentationState("sink", "model:1", { attemptId: "attempt-1" }); }); const restored = await deserializeContext(await serializeContext(context)); contextStorage.run(restored, () => { diff --git a/packages/eve/src/harness/instrumentation-state.ts b/packages/eve/src/harness/instrumentation-state.ts index c6a2a642ce..79817a7e2c 100644 --- a/packages/eve/src/harness/instrumentation-state.ts +++ b/packages/eve/src/harness/instrumentation-state.ts @@ -1,15 +1,37 @@ 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", { @@ -20,29 +42,61 @@ const InstrumentationStateKey = new ContextKey( }, ); -/** Keeps provider state from an interrupted step's discarded context changes. */ +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 { - const state = interrupted[InstrumentationStateKey.name]; - return state === undefined ? original : { ...original, [InstrumentationStateKey.name]: state }; + 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; } -/** One provider's durable state for one operation. */ +/** + * 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, - attemptId?: string, + owner: InstrumentationStateOwner = {}, ): InstrumentationStateLease { const key = stateKey(provider, idempotencyKey); let active = true; @@ -53,32 +107,39 @@ export function instrumentationStateSlot( active = false; }, set: (value) => { - if (!active || contextStorage.getStore() === undefined) return; - if (value === undefined) { - writeState((state) => writeSlot(state, key, undefined)); - return; - } - const record: InstrumentationStateRecord = { value: parseJsonValue(value) }; - const current = contextStorage.getStore()?.get(InstrumentationStateKey)?.[key]; - if (current?.abandoned === true) record.abandoned = true; - if (attemptId !== undefined) record.attemptId = attemptId; - writeState((state) => writeSlot(state, key, record)); + 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, - attemptId?: string, + owner: InstrumentationStateOwner = {}, ): void { - if (contextStorage.getStore() === undefined) return; const key = stateKey(provider, idempotencyKey); - writeState((state) => { + 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 }; - const owner = attemptId ?? current?.attemptId; - if (owner !== undefined) record.attemptId = owner; + assignOwner(record, resolvedOwner); if (current?.value !== undefined) record.value = current.value; return writeSlot(state, key, record); }); @@ -91,26 +152,35 @@ export function isInstrumentationStateAbandoned(provider: string, idempotencyKey ); } +/** + * 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; - writeState((state) => writeSlot(state, key, undefined)); + writeInstrumentationState((state) => writeSlot(state, key, undefined)); } -/** Releases children whose terminal may be omitted when an attempt ends. */ +/** 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) return; if ( + current === undefined || !Object.entries(current).some( ([key, record]) => key.startsWith(prefix) && record.attemptId === attemptId, ) ) { return; } - writeState((state) => { + writeInstrumentationState((state) => { const next = { ...state }; for (const [key, record] of Object.entries(state)) { if (key.startsWith(prefix) && record.attemptId === attemptId) delete next[key]; @@ -119,8 +189,93 @@ export function releaseInstrumentationAttemptState(provider: string, attemptId: }); } -function writeState(update: (state: InstrumentationStateMap) => InstrumentationStateMap): void { - loadContext().set(InstrumentationStateKey, (state) => update(state ?? {})); +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( @@ -128,12 +283,35 @@ function writeSlot( key: string, value: InstrumentationStateRecord | undefined, ): InstrumentationStateMap { - const next = { ...state }; - if (value === undefined) delete next[key]; - else next[key] = value; - return next; + 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}`; } @@ -144,11 +322,21 @@ function deserializeState(data: unknown): InstrumentationStateMap { 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 (typeof record["attemptId"] === "string") parsed.attemptId = record["attemptId"]; + 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; - if (parsed.abandoned === true || parsed.value !== undefined) state[key] = parsed; + 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 e2c3afd398..06ec26d7b9 100644 --- a/packages/eve/src/harness/tool-loop.test.ts +++ b/packages/eve/src/harness/tool-loop.test.ts @@ -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 { @@ -9632,36 +9631,6 @@ describe("createToolLoopHarness", () => { ); }); - 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", @@ -9692,7 +9661,7 @@ describe("createToolLoopHarness", () => { toolResults: [], }); const started = vi.fn(); - const hooks = createInstrumentationHooks([{ events: { "tool.call.started": started } }]); + const hooks = createInstrumentationHooks([{ events: { "action.started": started } }]); const { emit } = createEventCollector(); const runStep = createToolLoopHarness( createTestConfig("conversation", emit, { @@ -9707,8 +9676,8 @@ 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 ac027e7e19..23036f41df 100644 --- a/packages/eve/src/harness/tool-loop.ts +++ b/packages/eve/src/harness/tool-loop.ts @@ -113,11 +113,8 @@ 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"; @@ -580,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)), @@ -1042,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 @@ -1055,7 +1047,6 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn { attemptScope, instrumentationHooks, config.instrumentation?.runInContext, - resolveActionKind, ); const hooks = buildStepHooks({ diff --git a/packages/eve/src/public/instrumentation/provider.ts b/packages/eve/src/public/instrumentation/provider.ts index 47a0730af6..f92eaa0fae 100644 --- a/packages/eve/src/public/instrumentation/provider.ts +++ b/packages/eve/src/public/instrumentation/provider.ts @@ -15,7 +15,11 @@ 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, diff --git a/packages/eve/src/tracing/agent-otel-provider.test.ts b/packages/eve/src/tracing/agent-otel-provider.test.ts index ecd5d469d7..11d3fba36d 100644 --- a/packages/eve/src/tracing/agent-otel-provider.test.ts +++ b/packages/eve/src/tracing/agent-otel-provider.test.ts @@ -7,7 +7,7 @@ import { } from "@opentelemetry/sdk-trace-base"; import { describe, expect, it } from "vitest"; -import { createAiSdkHookBridge, type ActionKindResolver } from "#harness/ai-sdk-hook-bridge.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 { @@ -55,7 +55,6 @@ async function emitAttempt(input: { readonly hooks: InstrumentationHooks; readonly runInContext: InstrumentationContextRunner; readonly providerMetadata?: Readonly>; - readonly resolveActionKind?: ActionKindResolver; readonly sessionId: string; readonly skipModelTerminal?: boolean; readonly skipToolTerminal?: boolean; @@ -76,12 +75,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", @@ -273,7 +267,6 @@ describe("createAgentOtelInstrumentation", () => { const step = byName(spans, "agent.step")[0]!; const operation = byName(spans, "ai.streamText")[0]!; const model = byName(spans, "ai.streamText.doStream")[0]!; - const action = byName(spans, "agent.action")[0]!; const tool = byName(spans, "ai.toolCall")[0]!; const session = byName(spans, "agent.session")[0]!; @@ -287,8 +280,7 @@ describe("createAgentOtelInstrumentation", () => { expect(step.parentSpanContext?.spanId).toBe(turn.spanContext().spanId); expect(operation.parentSpanContext?.spanId).toBe(step.spanContext().spanId); expect(model.parentSpanContext?.spanId).toBe(operation.spanContext().spanId); - expect(action.parentSpanContext?.spanId).toBe(step.spanContext().spanId); - expect(tool.parentSpanContext?.spanId).toBe(action.spanContext().spanId); + expect(tool.parentSpanContext?.spanId).toBe(step.spanContext().spanId); expect(new Set(spans.map((span) => span.spanContext().traceId))).toHaveLength(1); expect(turn.events.map((event) => event.name)).toEqual([ "turn.started", @@ -310,12 +302,6 @@ describe("createAgentOtelInstrumentation", () => { "gen_ai.usage.cache_creation.input_tokens": 2, "gen_ai.usage.cache_read.input_tokens": 4, }); - expect(action.attributes).toMatchObject({ - "agent.action.kind": "tool-call", - "agent.action.name": "weather", - "agent.framework.name": "eve", - "agent.root.session.id": "session-1", - }); }); it("ends model and tool spans still open when the step attempt terminates", async () => { @@ -333,30 +319,11 @@ 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); }); - it("labels a subagent action by its kind, not as a plain tool", async () => { - const runtime = createRuntime(); - await emitAttempt({ - hooks: runtime.hooks, - resolveActionKind: () => "subagent-call", - runInContext: runtime.runInContext, - sessionId: "session-1", - turnId: "turn-1", - turnSequence: 0, - }); - await runtime.provider.forceFlush(); - - const action = runtime.exporter.getFinishedSpans().find((span) => span.name === "agent.action"); - expect(action?.attributes).toMatchObject({ - "agent.action.kind": "subagent-call", - "agent.action.name": "weather", - }); - }); - it("captures model and tool inputs/outputs on the operation spans", async () => { const runtime = createRuntime(); await emitAttempt({ @@ -391,10 +358,6 @@ 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"); }); it("truncates long conversations from the front, keeping valid JSON and recent messages", async () => { @@ -612,10 +575,8 @@ describe("createAgentOtelInstrumentation", () => { const replacementSpans = replacementRuntime.exporter.getFinishedSpans(); const turn = byName(replacementSpans, "agent.turn")[0]!; const step = byName(replacementSpans, "agent.step")[0]!; - const action = byName(replacementSpans, "agent.action")[0]!; expect(step.parentSpanContext?.spanId).toBe(turn.spanContext().spanId); - expect(action.parentSpanContext?.spanId).toBe(step.spanContext().spanId); expect(step.spanContext().traceId).toBe(turn.spanContext().traceId); }); @@ -832,7 +793,7 @@ describe("createAgentOtelInstrumentation", () => { expect(rolled.spanContext().traceId).not.toBe(parentWindow.spanContext().traceId); }); - it("marks a failed action without failing its turn", async () => { + it("marks a failed SDK tool call without failing its turn", async () => { const runtime = createRuntime(); await emitAttempt({ hooks: runtime.hooks, @@ -845,7 +806,7 @@ describe("createAgentOtelInstrumentation", () => { await runtime.provider.forceFlush(); const spans = runtime.exporter.getFinishedSpans(); - expect(byName(spans, "agent.action")[0]!.status.code).toBe(SpanStatusCode.ERROR); + expect(byName(spans, "ai.toolCall")[0]!.status.code).toBe(SpanStatusCode.ERROR); expect(byName(spans, "agent.turn")[0]!.status.code).toBe(SpanStatusCode.UNSET); }); }); diff --git a/packages/eve/src/tracing/agent-otel-provider.ts b/packages/eve/src/tracing/agent-otel-provider.ts index 9829f205d8..4eb4655ee6 100644 --- a/packages/eve/src/tracing/agent-otel-provider.ts +++ b/packages/eve/src/tracing/agent-otel-provider.ts @@ -53,9 +53,7 @@ interface AttemptSpanState { readonly step: SpanState; } -interface ToolSpanState extends SpanState { - readonly toolSpan: Span; -} +type ToolSpanState = SpanState; export interface AgentOtelInstrumentationInput { /** @@ -345,26 +343,7 @@ export function createAgentOtelInstrumentation( const onToolCallStarted = (event: InstrumentationToolCallStartedEvent): void => { 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 span = input.tracer.startSpan( "ai.toolCall", { attributes: { @@ -373,16 +352,15 @@ export function createAgentOtelInstrumentation( "gen_ai.tool.name": event.toolName, }, }, - actionContext, + attempt.step.context, ); 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, + context: trace.setSpan(attempt.step.context, span), + span, }; getExecutionContexts(event.scope).tools.set(event.idempotencyKey, state.context); getSpanStates(toolSpans, event.scope).set(event.idempotencyKey, state); @@ -393,16 +371,13 @@ export function createAgentOtelInstrumentation( 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(); }; @@ -543,10 +518,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/local-instrumentation-runtime.scenario.test.ts b/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts index e93c27f689..97c9180560 100644 --- a/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts +++ b/packages/eve/src/tracing/local-instrumentation-runtime.scenario.test.ts @@ -157,7 +157,6 @@ describe("local instrumentation runtime", () => { "agent.step", "ai.streamText", "ai.streamText.doStream", - "agent.action", "ai.toolCall", "user.model-work", "user.tool-work", @@ -171,8 +170,7 @@ describe("local instrumentation runtime", () => { expect(span(spans, "user.model-work").parentSpanId).toBe( span(spans, "ai.streamText.doStream").spanId, ); - expect(span(spans, "agent.action").parentSpanId).toBe(span(spans, "agent.step").spanId); - expect(span(spans, "ai.toolCall").parentSpanId).toBe(span(spans, "agent.action").spanId); + expect(span(spans, "ai.toolCall").parentSpanId).toBe(span(spans, "agent.step").spanId); expect(span(spans, "user.tool-work").parentSpanId).toBe(span(spans, "ai.toolCall").spanId); const listed = await listLocalTraces(appRoot); expect(listed).toHaveLength(1);