From 53e22d094dace7f7450ac086fad9709c653a62cb Mon Sep 17 00:00:00 2001 From: Nicolas Molina Date: Thu, 27 Aug 2026 22:36:57 -0300 Subject: [PATCH] fix(eve): dispatch subagent called hooks Signed-off-by: Nicolas Molina --- .changeset/restore-subagent-called-hooks.md | 5 + docs/guides/hooks.md | 4 +- .../dispatch-runtime-actions-shared.ts | 100 ++++++++----- ...h-runtime-actions-step.integration.test.ts | 138 +++++++++++++++--- .../dispatch-runtime-actions-step.ts | 5 +- .../tasks/parent/dispatch-task-step.ts | 5 +- 6 files changed, 188 insertions(+), 69 deletions(-) create mode 100644 .changeset/restore-subagent-called-hooks.md diff --git a/.changeset/restore-subagent-called-hooks.md b/.changeset/restore-subagent-called-hooks.md new file mode 100644 index 0000000000..7a7f0d9766 --- /dev/null +++ b/.changeset/restore-subagent-called-hooks.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Deliver parent `subagent.called` events to authored hooks in both plain and task dispatch modes. Hook failures after a child starts are logged without replaying the child dispatch. diff --git a/docs/guides/hooks.md b/docs/guides/hooks.md index c169d92e37..4b73d3ee05 100644 --- a/docs/guides/hooks.md +++ b/docs/guides/hooks.md @@ -26,7 +26,7 @@ The slug is the path-relative basename. `agent/hooks/audit.ts` becomes `"audit"` `defineHook`, `HookDefinition`, and `HookContext` live on `eve/hooks`. -A hook file declares stream-event subscribers under the `events` map, keyed by event type, with `*` matching every event. Subscribe to any event in the runtime stream vocabulary documented in [Sessions, runs and streaming](../concepts/sessions-runs-and-streaming), including the lifecycle events `session.started`, `turn.completed`, `message.completed`, `action.partial`, and `action.result`. Handlers are observe-only. They cannot inject model context. To contribute runtime model messages, use `defineDynamic` and `defineInstructions` in `agent/instructions/`. +A hook file declares stream-event subscribers under the `events` map, keyed by event type, with `*` matching every event. Subscribe to any event in the runtime stream vocabulary documented in [Sessions, runs and streaming](../concepts/sessions-runs-and-streaming), including the lifecycle events `session.started`, `turn.completed`, `message.completed`, `action.partial`, `action.result`, and `subagent.called`. Handlers are observe-only. They cannot inject model context. To contribute runtime model messages, use `defineDynamic` and `defineInstructions` in `agent/instructions/`. ## Scope side effects to a channel @@ -178,6 +178,8 @@ Hooks always run after the event is durably recorded, so if a hook throws, the s A thrown handler propagates through the emit composer and surfaces as `turn.failed`. If a hook subscribed to a failure-cascade event also throws, it escalates to `session.failed`. For belt-and-suspenders semantics inside a hook, wrap the body in `try`/`catch`. eve treats a thrown hook as a real failure. +`subagent.called` is the exception. The child is already running when eve emits this event, so a hook failure is logged instead of propagating and replaying the dispatch. The parent stream keeps the `subagent.called` event that was written before the hook ran. + ## Subagent isolation Subagents may carry their own `agent/hooks/` directory. Subagent hooks fire only inside the subagent scope. Parent-agent hooks do not fire for subagent turns, and subagent hooks see only the subagent's own context. diff --git a/packages/eve/src/execution/dispatch-runtime-actions-shared.ts b/packages/eve/src/execution/dispatch-runtime-actions-shared.ts index a600eb934f..700da0f2cd 100644 --- a/packages/eve/src/execution/dispatch-runtime-actions-shared.ts +++ b/packages/eve/src/execution/dispatch-runtime-actions-shared.ts @@ -21,6 +21,7 @@ import { SandboxKey, } from "#context/keys.js"; import { type AlsContext, ContextContainer } from "#context/container.js"; +import { dispatchStreamEventHooks } from "#context/hook-lifecycle.js"; import { withContextScope } from "#context/run-step.js"; import { BundleKey, @@ -70,6 +71,7 @@ import { resolveSubagentDepth } from "#harness/subagent-depth.js"; import { getDynamicSubagentSelection } from "#context/dynamic-subagent-lifecycle.js"; import { resolveEffectiveAgentRuntime } from "#execution/effective-agent-config.js"; import { isTaskControlAction } from "#execution/tasks/parent/dispatch.js"; +import { setChannelContext } from "#execution/channel-context.js"; const log = createLogger("execution.dispatch-runtime-actions"); @@ -145,6 +147,7 @@ export interface PreparedRuntimeActionDispatch { readonly bundle: CompiledBundle; readonly capabilities: Parameters[0]["capabilities"]; readonly channelMetadata: Parameters[0]["channelMetadata"]; + readonly ctx: ContextContainer; /** * Number of freshly started local subagents in the plan. The parent's * remaining token quota is split across these, the children that @@ -276,6 +279,7 @@ async function prepareActionDispatch(input: { bundle, capabilities: ctx.get(CapabilitiesKey), channelMetadata: ctx.get(ChannelInstrumentationKey), + ctx, fanoutSize: input.fanoutSize ?? plan.filter((entry) => entry.kind === "start" && entry.target.kind === "local").length, @@ -323,57 +327,73 @@ export async function emitSubagentCalled(input: { readonly adapter: ChannelAdapter; readonly adapterCtx: ChannelAdapterContext; readonly batchEvent: { readonly sequence: number; readonly turnId: string }; + readonly ctx: ContextContainer; readonly entry: Extract; readonly outcome: Extract; - readonly sessionId: string; + readonly session: RuntimeSession; readonly writer: WritableStreamDefaultWriter; -}): Promise { +}): Promise { const { entry, outcome } = input; try { - const action = entry.kind === "resume" ? entry.action : entry.target.action; - const dynamicRemoteAgent = - entry.kind === "resume" - ? entry.dynamicRemoteAgent - : entry.target.kind === "remote" - ? entry.target.dynamicRemoteAgent - : undefined; - const parentEvent = await callAdapterEventHandler( - input.adapter, - createSubagentCalledEvent({ - callId: outcome.callId, - childSessionId: outcome.address.sessionId, - name: outcome.name, - remote: - outcome.address.kind === "agent/remote" - ? { - // The proxy route re-resolves outbound auth from this key via - // resolveRemoteAgentStreamHeaders: a node id lands in - // subagentRegistry.subagentsByNodeId (static definition), a - // credentialsStepId lands in the step registry (dynamic - // definition). Both sides of this ternary must stay in sync - // with that lookup order. - resolverId: - dynamicRemoteAgent === undefined - ? action.nodeId - : dynamicRemoteAgent.credentialsStepId, - url: outcome.address.url, - } - : undefined, - sequence: input.batchEvent.sequence, - sessionId: input.sessionId, - toolName: outcome.toolName, - turnId: input.batchEvent.turnId, - workflowId: workflowEntryReference.workflowId, - }), - input.adapterCtx, - ); - await input.writer.write(encodeMessageStreamEvent(stampMessageStreamEvent(parentEvent))); + const scoped = await withContextScope(input.ctx, input.session, async (enrichedSession) => { + const action = entry.kind === "resume" ? entry.action : entry.target.action; + const dynamicRemoteAgent = + entry.kind === "resume" + ? entry.dynamicRemoteAgent + : entry.target.kind === "remote" + ? entry.target.dynamicRemoteAgent + : undefined; + const parentEvent = await callAdapterEventHandler( + input.adapter, + createSubagentCalledEvent({ + callId: outcome.callId, + childSessionId: outcome.address.sessionId, + name: outcome.name, + remote: + outcome.address.kind === "agent/remote" + ? { + // The proxy route re-resolves outbound auth from this key via + // resolveRemoteAgentStreamHeaders: a node id lands in + // subagentRegistry.subagentsByNodeId (static definition), a + // credentialsStepId lands in the step registry (dynamic + // definition). Both sides of this ternary must stay in sync + // with that lookup order. + resolverId: + dynamicRemoteAgent === undefined + ? action.nodeId + : dynamicRemoteAgent.credentialsStepId, + url: outcome.address.url, + } + : undefined, + sequence: input.batchEvent.sequence, + sessionId: input.session.sessionId, + toolName: outcome.toolName, + turnId: input.batchEvent.turnId, + workflowId: workflowEntryReference.workflowId, + }), + input.adapterCtx, + ); + setChannelContext(input.ctx, { + ...input.adapter, + state: { ...input.adapterCtx.state }, + }); + const stamped = stampMessageStreamEvent(parentEvent); + await input.writer.write(encodeMessageStreamEvent(stamped)); + await dispatchStreamEventHooks({ + ctx: input.ctx, + event: stamped, + registry: input.ctx.require(BundleKey).hookRegistry, + }); + return { result: undefined, session: enrichedSession }; + }); + return scoped.session; } catch (error) { logError(log, "subagent.called emission failed", error, { callId: outcome.callId, childSessionId: outcome.address.sessionId, toolName: outcome.toolName, }); + return input.session; } } diff --git a/packages/eve/src/execution/dispatch-runtime-actions-step.integration.test.ts b/packages/eve/src/execution/dispatch-runtime-actions-step.integration.test.ts index 8e24a3cba7..cacda71c2b 100644 --- a/packages/eve/src/execution/dispatch-runtime-actions-step.integration.test.ts +++ b/packages/eve/src/execution/dispatch-runtime-actions-step.integration.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { ChannelAdapter } from "#channel/adapter.js"; import type { SessionAuthContext } from "#channel/types.js"; import { ContextContainer, loadContext } from "#context/container.js"; +import type { HookContext, HookDefinition, HookEventMap } from "#public/definitions/hook.js"; import { RemoteAgentContinueRequestError } from "#execution/remote-agent-dispatch.js"; import { RuntimeSessionOwnershipConflictError } from "#execution/runtime-errors.js"; import type { DurableSessionState } from "#execution/durable-session-store.js"; @@ -24,14 +25,7 @@ import type { HarnessSession } from "#harness/types.js"; import { getSessionTaskIndex } from "#tasks/session-index.js"; import { recordSessionTask } from "#tasks/session-index.js"; import * as taskRunControl from "#execution/tasks/parent/run-parent.js"; -import { - AuthKey, - CapabilitiesKey, - ChannelInstrumentationKey, - InitiatorAuthKey, - SessionIdKey, - SessionKey, -} from "#context/keys.js"; +import { AuthKey, InitiatorAuthKey, SessionIdKey, SessionKey } from "#context/keys.js"; import { BundleKey, ChannelKey } from "#runtime/sessions/runtime-context-keys.js"; import { createBundledRuntimeCompiledArtifactsSource } from "#runtime/compiled-artifacts-source.js"; import type { @@ -41,6 +35,7 @@ import type { import type { RuntimeSandboxRegistry } from "#runtime/sandbox/registry.js"; import type { ResolvedSandboxDefinition } from "#runtime/types.js"; import { mockSandbox } from "#internal/testing/mocks/mock-sandbox.js"; +import { createRuntimeHookRegistry } from "#runtime/hooks/registry.js"; const mocks = vi.hoisted(() => ({ continueRemoteAgentSession: vi.fn(), @@ -306,6 +301,82 @@ describe("dispatchRuntimeActionsStep child starts", () => { expect(writes).toHaveLength(1); }); + it.each([ + ["plain", dispatchRuntimeActionsStep], + ["task", dispatchTaskStep], + ] as const)( + "delivers subagent.called to the parent stream and typed hooks in %s mode", + async (_mode, dispatch) => { + const session = createStartSession({ kind: "local" }); + const hookedEvents: HookEventMap["subagent.called"][] = []; + installContext( + session, + undefined, + dispatch === dispatchTaskStep, + null, + async (event, ctx) => { + expect(ctx.session.id).toBe("parent-session"); + hookedEvents.push(event); + }, + ); + const writes: Uint8Array[] = []; + if (dispatch === dispatchTaskStep) { + vi.spyOn(taskRunControl, "sendTaskCommandToOwner").mockResolvedValue({ + runId: "task-run-1", + }); + } + + const result = await dispatch({ + parentContinuationToken: "turn-inbox", + parentWritable: createWritable(writes), + serializedContext: {}, + sessionState: BASE_STATE, + }); + + const streamedEvents = writes.map(decodeEvent); + expect(streamedEvents).toHaveLength(1); + expect(streamedEvents[0]).toMatchObject({ type: "subagent.called" }); + expect(hookedEvents).toHaveLength(1); + expect(hookedEvents[0]).toEqual(streamedEvents[0]); + expect(hookedEvents[0]?.meta.id).toBe(streamedEvents[0]?.meta.id); + expect(result.sessionState.snapshot?.session.sandboxState).toEqual({ + initialized: false, + session: null, + }); + }, + ); + + it("keeps a started child and its stream event when a subagent.called hook throws", async () => { + const session = createStartSession({ kind: "local" }); + installContext(session, undefined, false, null, async () => { + throw new Error("subagent hook failed"); + }); + const writes: Uint8Array[] = []; + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); + + const result = await dispatchRuntimeActionsStep({ + parentContinuationToken: "turn-inbox", + parentWritable: createWritable(writes), + serializedContext: {}, + sessionState: BASE_STATE, + }); + + expect(writes.map(decodeEvent)).toEqual([expect.objectContaining({ type: "subagent.called" })]); + expect(errorSpy).toHaveBeenCalledWith( + expect.stringContaining("subagent.called emission failed"), + expect.objectContaining({ callId: "call-1", childSessionId: CHILD_SESSION_ID }), + ); + expect(mocks.createSession).toHaveBeenCalledTimes(1); + expect(getAgentHandleStore(readResultSessionState(result, session))).toEqual({ + handles: [ + expect.objectContaining({ + address: expect.objectContaining({ sessionId: CHILD_SESSION_ID }), + phase: "running", + }), + ], + }); + }); + it("opens a shared parent sandbox with session context and durable backend tags", async () => { const session = createStartSession({ kind: "local" }); const observedSessionIds: string[] = []; @@ -1274,13 +1345,36 @@ function installContext( remote?: { readonly definition: unknown; readonly nodeId: string }, tasks = false, auth: SessionAuthContext | null = null, + onSubagentCalled?: (event: HookEventMap["subagent.called"], ctx: HookContext) => Promise, ): void { const subagentsByNodeId = new Map(); if (remote !== undefined) { subagentsByNodeId.set(remote.nodeId, { definition: remote.definition }); } + const typedHook: HookDefinition<"subagent.called"> = { + events: onSubagentCalled === undefined ? {} : { "subagent.called": onSubagentCalled }, + }; + const hookRegistry = createRuntimeHookRegistry([ + { + events: typedHook.events as never, + exportName: undefined, + logicalPath: "hooks/subagent-called.ts", + slug: "subagent-called", + sourceId: "hooks/subagent-called.ts", + sourceKind: "module", + }, + ]); + const root = { + agent: { config: { name: "test-agent" }, connections: [] }, + nodeId: "__root__", + sandboxRegistry: { sandbox: null }, + turnAgent: session.agent, + }; const bundle = { - compiledArtifactsSource: {}, + compiledArtifactsSource: createBundledRuntimeCompiledArtifactsSource(), + graph: { nodesByNodeId: new Map([["__root__", root]]), root }, + hookRegistry, + nodeId: "__root__", resolvedAgent: { config: tasks ? { experimental: { tasks: true } } : {}, }, @@ -1294,21 +1388,13 @@ function installContext( workspaceSpec: {}, }, }; - const values = new Map([ - [AuthKey, auth], - [BundleKey, bundle], - [CapabilitiesKey, undefined], - [ChannelInstrumentationKey, undefined], - [InitiatorAuthKey, null], - [ChannelKey, ADAPTER], - ]); - mocks.deserializeContext.mockResolvedValue({ - get: (key: unknown) => values.get(key), - require: (key: unknown) => { - if (!values.has(key)) throw new Error("missing context key"); - return values.get(key); - }, - }); + const ctx = new ContextContainer(); + ctx.set(AuthKey, auth); + ctx.set(BundleKey, bundle as never); + ctx.set(InitiatorAuthKey, null); + ctx.set(SessionIdKey, session.sessionId); + ctx.set(ChannelKey, ADAPTER); + mocks.deserializeContext.mockResolvedValue(ctx); mocks.readDurableSession.mockResolvedValue(session); } @@ -1406,3 +1492,7 @@ function createWritable(writes: Uint8Array[] = []): WritableStream { }, }); } + +function decodeEvent(chunk: Uint8Array): HookEventMap["subagent.called"] { + return JSON.parse(new TextDecoder().decode(chunk).trim()) as HookEventMap["subagent.called"]; +} diff --git a/packages/eve/src/execution/dispatch-runtime-actions-step.ts b/packages/eve/src/execution/dispatch-runtime-actions-step.ts index 287dbc27d6..e9bffcc4f8 100644 --- a/packages/eve/src/execution/dispatch-runtime-actions-step.ts +++ b/packages/eve/src/execution/dispatch-runtime-actions-step.ts @@ -105,13 +105,14 @@ export async function dispatchRuntimeActionsStep( continue; } - await emitSubagentCalled({ + nextSession = await emitSubagentCalled({ adapter: prepared.adapter, adapterCtx: prepared.adapterCtx, batchEvent: batch.event, + ctx: prepared.ctx, entry, outcome, - sessionId: session.sessionId, + session: nextSession, writer, }); } diff --git a/packages/eve/src/execution/tasks/parent/dispatch-task-step.ts b/packages/eve/src/execution/tasks/parent/dispatch-task-step.ts index 7b3048afa4..6eeab40730 100644 --- a/packages/eve/src/execution/tasks/parent/dispatch-task-step.ts +++ b/packages/eve/src/execution/tasks/parent/dispatch-task-step.ts @@ -193,13 +193,14 @@ export async function dispatchTaskStep( } pendingTasks.push(delegated); - await emitSubagentCalled({ + nextSession = await emitSubagentCalled({ adapter: prepared.adapter, adapterCtx: prepared.adapterCtx, batchEvent: batch.event, + ctx: prepared.ctx, entry, outcome, - sessionId: session.sessionId, + session: nextSession, writer, }); }