fix(eve): deliver subagent.called events to hooks

.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.

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.

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<typeof buildSubagentRunInput>[0]["capabilities"];
   readonly channelMetadata: Parameters<typeof buildSubagentRunInput>[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<DispatchPlanEntry, { readonly kind: "resume" | "start" }>;
   readonly outcome: Extract<DispatchOutcome, { readonly kind: "called" }>;
-  readonly sessionId: string;
+  readonly session: RuntimeSession;
   readonly writer: WritableStreamDefaultWriter<Uint8Array>;
-}): Promise<void> {
+}): Promise<RuntimeSession> {
   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;
   }
 }
 

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>,
 ): void {
   const subagentsByNodeId = new Map<string, { definition: unknown }>();
   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<unknown, unknown>([
-    [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);
 }
 
@@ -1407,3 +1493,7 @@ function createWritable(writes: Uint8Array[] = []): WritableStream<Uint8Array> {
     },
   });
 }
+
+function decodeEvent(chunk: Uint8Array): HookEventMap["subagent.called"] {
+  return JSON.parse(new TextDecoder().decode(chunk).trim()) as HookEventMap["subagent.called"];
+}

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,
       });
     }

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,
       });
     }