fix(eve): execution - recover hook failures and preserve cancellation notifications (#3091)

Signed-off-by: Andrew Barba <barba@hey.com>

.changeset/preserve-cancelled-task-notification.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Preserve a cancelled task's notification to its parent when shutdown exceeds the cooperative grace period. Parents no longer wait for a notification from a task that was already cancelled.

.changeset/recover-boundary-hook-failures.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Keep conversation sessions available after a `turn.started` or `step.started` handler throws. The failed turn reports the error, and a later message can resume the same session.

docs/guides/hooks.md

@@ -176,7 +176,7 @@ Hooks always run after the event is durably recorded, so if a hook throws, the s
 
 ## What happens when a hook throws
 
-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.
+A thrown handler propagates through the emit composer and surfaces as `turn.failed`. In a conversation session, this includes handlers for `turn.started` and the first `step.started` of a model call: the failed turn ends with `session.waiting`, and the next message can start another turn. Task-mode boundary failures remain terminal. 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 isolation
 

docs/subagents/index.mdx

@@ -171,7 +171,7 @@ Do not rely on subagent delegation by itself as an approval boundary. Put sensit
 
 Each delegated subagent spins up its own child session and stream. The parent stream carries the control-plane events `subagent.called` and `subagent.completed`, plus interactive `input.requested`, `authorization.required`, and `authorization.completed` events proxied from descendants so the root channel can prompt the user. To follow the child's other progress, read `subagent.called.data.childSessionId` and subscribe at `GET /eve/v1/session/:childSessionId/stream`.
 
-A background task that was already admitted survives cancellation of the turn that started it; background work that has not yet been admitted is rejected with the cancelled step. Use `task_cancel` to stop an admitted task. Parent-session finalization cancels remaining live tasks.
+A background task that was already admitted survives cancellation of the turn that started it; background work that has not yet been admitted is rejected with the cancelled step. Use `task_cancel` to stop an admitted task. Cancellation delivers the task's final notification to an active parent even if the task must be stopped forcibly. Parent-session finalization cancels remaining live tasks.
 
 Subagent model calls automatically retry classified transient provider failures, including overload errors delivered after a stream starts. eve makes at most three fresh model-call attempts, repeating only the current uncommitted call so completed earlier steps, tool results, and sandbox work remain available to the child. Other recoverable task errors fall back to Workflow's durable step retry from the last committed session snapshot. Exhausting the transient model-call attempts or the dedicated empty-response reissue returns one failed task result instead of stacking both retry budgets; terminal errors fail immediately.
 

e2e/fixtures/agent-basic-runtime/agent/channels/eve.ts

@@ -20,6 +20,9 @@ const authenticateA: AuthFn<Request> = (request) =>
   request.headers.get("authorization") === PRINCIPAL_A ? principal("issuer-a") : null;
 const authenticateB: AuthFn<Request> = (request) =>
   request.headers.get("authorization") === PRINCIPAL_B ? principal("issuer-b") : null;
-const authenticateEvalDriver: AuthFn<Request> = () => principal("eval-driver");
+const authenticateEvalDriver: AuthFn<Request> = (request) => ({
+  ...principal("eval-driver"),
+  attributes: { denyBoundary: request.headers.get("x-e2e-deny-boundary") ?? "" },
+});
 
 export default eveChannel({ auth: [authenticateA, authenticateB, authenticateEvalDriver] });

e2e/fixtures/agent-basic-runtime/agent/hooks/admission.ts

@@ -0,0 +1,14 @@
+import { defineHook } from "eve/hooks";
+
+export default defineHook({
+  events: {
+    "*"(event, ctx) {
+      if (
+        (event.type === "turn.started" || event.type === "step.started") &&
+        ctx.session.auth.current?.attributes.denyBoundary === event.type
+      ) {
+        throw new Error("Fixture admission denied.");
+      }
+    },
+  },
+});

e2e/fixtures/agent-basic-runtime/evals/runtime/boundary-hook-failure.eval.ts

@@ -0,0 +1,20 @@
+import { defineEval } from "eve/evals";
+
+export default defineEval({
+  description: "Boundary hook failures end one turn while the conversation remains resumable.",
+  async test(t) {
+    for (const boundary of ["turn.started", "step.started"]) {
+      const session = t.newSession();
+      const failed = await session.send("Deny this turn.", {
+        headers: { "x-e2e-deny-boundary": boundary },
+      });
+      failed.event("turn.failed", { count: 1, data: { code: "EVENT_HANDLER_FAILED" } });
+      failed.event("session.waiting", { count: 1 });
+      failed.notEvent("session.failed");
+      const recovered = await session.send("Reply with ready.");
+      recovered.expectOk();
+      recovered.event("turn.started", { count: 1, data: { sequence: 1 } });
+      recovered.notEvent("session.started");
+    }
+  },
+});

packages/eve/src/context/hook-lifecycle.integration.test.ts

@@ -2,7 +2,11 @@ import { describe, expect, it } from "vitest";
 
 import { createRuntimeHookRegistry } from "#runtime/hooks/registry.js";
 import type { ResolvedHookDefinition } from "#runtime/types.js";
-import type { UnstampedMessageStreamEvent } from "#protocol/message.js";
+import {
+  createStepStartedEvent,
+  createTurnStartedEvent,
+  type UnstampedMessageStreamEvent,
+} from "#protocol/message.js";
 import { stampTestEvent } from "#internal/testing/events.js";
 import { mockSandbox } from "#internal/testing/mocks/mock-sandbox.js";
 import { ContextContainer, contextStorage } from "./container.js";
@@ -111,6 +115,41 @@ describe("dispatchStreamEventHooks", () => {
     ).rejects.toThrow(/event hook boom/);
   });
 
+  it.each(["turn.started", "step.started"] as const)(
+    "identifies an authored %s rejection for turn recovery",
+    async (type) => {
+      const cause = new Error("admission denied");
+      const registry = createRuntimeHookRegistry([
+        hook("admission", {
+          events: {
+            "*": async () => {
+              throw cause;
+            },
+          },
+        }),
+      ]);
+      const ctx = buildCtx();
+      await expect(
+        contextStorage.run(ctx, () =>
+          dispatchStreamEventHooks({
+            ctx,
+            registry,
+            event: stampTestEvent(
+              type === "turn.started"
+                ? createTurnStartedEvent({ sequence: 0, turnId: "turn_0" })
+                : createStepStartedEvent({
+                    sequence: 0,
+                    turnId: "turn_0",
+                    stepIndex: 0,
+                    modelId: "test",
+                  }),
+            ),
+          }),
+        ),
+      ).rejects.toMatchObject({ name: "BoundaryHookError", cause, message: "admission denied" });
+    },
+  );
+
   it("can delete the runtime sandbox from a session.completed hook", async () => {
     let deletions = 0;
     const sandbox = mockSandbox({

packages/eve/src/context/hook-lifecycle.ts

@@ -1,3 +1,4 @@
+import { BoundaryHookError } from "#shared/boundary-hook-error.js";
 import { getAdapterKind } from "#channel/adapter.js";
 import type { MessageStreamEvent } from "#protocol/message.js";
 import type { HookContext } from "#public/definitions/hook.js";
@@ -26,11 +27,18 @@ export async function dispatchStreamEventHooks(input: {
   }
 
   const hookCtx = buildHookContext(input.ctx);
-  for (const entry of typed) {
-    await entry.handler(input.event, hookCtx);
-  }
-  for (const entry of wildcard) {
-    await entry.handler(input.event, hookCtx);
+  try {
+    for (const entry of typed) {
+      await entry.handler(input.event, hookCtx);
+    }
+    for (const entry of wildcard) {
+      await entry.handler(input.event, hookCtx);
+    }
+  } catch (error) {
+    if (input.event.type === "turn.started" || input.event.type === "step.started") {
+      throw new BoundaryHookError(error);
+    }
+    throw error;
   }
 }
 

packages/eve/src/execution/tasks/parent/cancel-notification.integration.test.ts

@@ -0,0 +1,30 @@
+import { describe, expect, it } from "vitest";
+
+import { createTestRuntime } from "#internal/testing/app-harness.js";
+import { taskCancelNotificationWorkflow } from "#internal/testing/task-cancel-notification-workflow.js";
+import { start } from "#internal/workflow/runtime.js";
+
+describe("task cancellation parent notification", () => {
+  it("delivers the committed view from the parent step after forcing a slow lifecycle to stop", async () => {
+    const runtime = await createTestRuntime({ agent: { name: "task-cancel-notification" } });
+    await runtime.run(async () => {
+      const run = await start(taskCancelNotificationWorkflow, []);
+      try {
+        const result = await run.returnValue;
+        expect(result.taskRunStatus).toBe("cancelled");
+        expect(result.view.status).toBe("cancelled");
+        expect(result.notification).toMatchObject({
+          kind: "send",
+          payload: {
+            message: `Background task ${result.view.taskId} (slow-cancel) is cancelled.`,
+            task: { views: [result.view] },
+          },
+          taskDeliveryId: `${result.view.taskId}:ready:cancelled`,
+        });
+      } finally {
+        const status = await run.status;
+        if (status === "pending" || status === "running") await run.cancel();
+      }
+    });
+  }, 30_000);
+});

packages/eve/src/execution/tasks/parent/dispatch.test.ts

@@ -3,6 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { cancelOwnedTask, executeTaskControlAction } from "#execution/tasks/parent/dispatch.js";
 import { readLatestTaskView, sendTaskCommand } from "#execution/tasks/parent/run-parent.js";
 import { cancelWorkflowToolRun } from "#execution/tools/workflow/cancel.js";
+import { resumeSessionInbox } from "#execution/wire/session-inbox-resume.js";
 
 const { cancelRun, getRun } = vi.hoisted(() => ({
   cancelRun: vi.fn(),
@@ -14,6 +15,7 @@ vi.mock("#execution/tasks/parent/run-parent.js", () => ({
   sendTaskCommand: vi.fn(),
 }));
 vi.mock("#execution/tools/workflow/cancel.js", () => ({ cancelWorkflowToolRun: vi.fn() }));
+vi.mock("#execution/wire/session-inbox-resume.js", () => ({ resumeSessionInbox: vi.fn() }));
 vi.mock("#internal/workflow/runtime.js", () => ({
   cancelRun,
   getRun,
@@ -60,6 +62,7 @@ describe("task cancellation", () => {
       "Task task-1 was cancelled.",
     );
     expect(cancelRun).not.toHaveBeenCalled();
+    expect(resumeSessionInbox).not.toHaveBeenCalled();
   });
 
   it("does not reinterpret an unknown executor binding", async () => {
@@ -86,15 +89,23 @@ describe("task cancellation", () => {
       .fn()
       .mockRejectedValueOnce(new Error("Child cancellation failed"))
       .mockResolvedValueOnce(undefined);
-    await expect(cancelOwnedTask({ cancelOwnedWork, entry })).rejects.toThrow(
+    const session = { sessionId: "parent-session" } as Parameters<
+      typeof cancelOwnedTask
+    >[0]["session"];
+    await expect(cancelOwnedTask({ cancelOwnedWork, entry, session })).rejects.toThrow(
       "Child cancellation failed",
     );
+    expect(resumeSessionInbox).not.toHaveBeenCalled();
 
     vi.mocked(sendTaskCommand).mockResolvedValue("unreachable");
-    await expect(cancelOwnedTask({ cancelOwnedWork, entry })).resolves.toMatchObject({
+    await expect(cancelOwnedTask({ cancelOwnedWork, entry, session })).resolves.toMatchObject({
       status: "cancelled",
     });
     expect(cancelOwnedWork).toHaveBeenCalledTimes(2);
+    expect(resumeSessionInbox).toHaveBeenCalledTimes(1);
+    expect(cancelOwnedWork.mock.invocationCallOrder[1]).toBeLessThan(
+      vi.mocked(resumeSessionInbox).mock.invocationCallOrder[0]!,
+    );
   });
 
   it("leaves child work untouched when completion won the cancellation race", async () => {
@@ -126,6 +137,59 @@ describe("task cancellation", () => {
       cancelReason: "Task task-1 was cancelled.",
     });
   });
+
+  it("preserves the committed parent notification when cancellation stops a slow task run", async () => {
+    const view = { metadata: entry.metadata, status: "cancelled", taskId: entry.taskId } as const;
+    vi.mocked(readLatestTaskView).mockResolvedValue(view);
+    getRun.mockReturnValue({ status: Promise.resolve("running") });
+    const session = { sessionId: "parent-session" } as Parameters<
+      typeof cancelOwnedTask
+    >[0]["session"];
+    const cancelled = cancelOwnedTask({ entry, session });
+
+    await vi.advanceTimersByTimeAsync(999);
+    expect(cancelRun).not.toHaveBeenCalled();
+    expect(resumeSessionInbox).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(1);
+    await expect(cancelled).resolves.toEqual(view);
+
+    expect(cancelRun).toHaveBeenCalledTimes(1);
+    expect(resumeSessionInbox).toHaveBeenCalledExactlyOnceWith("eve:session:parent-session:inbox", {
+      kind: "send",
+      payload: {
+        message: "Background task task-1 (export) is cancelled.",
+        task: { views: [view] },
+      },
+      taskDeliveryId: "task-1:ready:cancelled",
+    });
+    expect(vi.mocked(cancelRun).mock.invocationCallOrder[0]).toBeLessThan(
+      vi.mocked(resumeSessionInbox).mock.invocationCallOrder[0]!,
+    );
+  });
+
+  it("retries the parent notification after the cancelled task inbox is gone", async () => {
+    const view = { metadata: entry.metadata, status: "cancelled", taskId: entry.taskId } as const;
+    vi.mocked(readLatestTaskView).mockResolvedValue(view);
+    getRun.mockReturnValue({ status: Promise.resolve("running") });
+    vi.mocked(resumeSessionInbox).mockRejectedValueOnce(new Error("temporary delivery failure"));
+    const session = { sessionId: "parent-session" } as Parameters<
+      typeof cancelOwnedTask
+    >[0]["session"];
+    const cancelled = cancelOwnedTask({ entry, session });
+    const failed = expect(cancelled).rejects.toThrow("temporary delivery failure");
+    await vi.runAllTimersAsync();
+    await failed;
+
+    vi.mocked(sendTaskCommand).mockResolvedValue("unreachable");
+    getRun.mockReturnValue({ status: Promise.resolve("cancelled") });
+    await expect(cancelOwnedTask({ entry, session })).resolves.toEqual(view);
+    expect(cancelRun).toHaveBeenCalledTimes(1);
+    expect(cancelWorkflowToolRun).toHaveBeenCalledTimes(2);
+    expect(resumeSessionInbox).toHaveBeenCalledTimes(2);
+    expect(vi.mocked(resumeSessionInbox).mock.calls[1]).toEqual(
+      vi.mocked(resumeSessionInbox).mock.calls[0],
+    );
+  });
 });
 
 describe("task updates", () => {

packages/eve/src/execution/tasks/parent/dispatch.ts

@@ -9,6 +9,8 @@ import {
 } from "#execution/tasks/parent/control-shared.js";
 import type { BackgroundTask } from "#execution/tasks/parent/delegate.js";
 import { sendTaskCommand } from "#execution/tasks/parent/run-parent.js";
+import { wakeTaskParentStep } from "#execution/tasks/child/steps.js";
+import { sessionCommandHookToken } from "#execution/session-command-token.js";
 import {
   cancelTaskOwnedWork,
   type TaskExecutorCancel,
@@ -123,7 +125,7 @@ export async function cancelOwnedTask(input: {
   readonly serializedContext?: Record<string, unknown>;
   readonly session?: RuntimeSession;
 }): Promise<TaskView> {
-  await sendTaskCommand({
+  const delivery = await sendTaskCommand({
     command: { kind: "cancel" },
     taskInboxToken: input.entry.taskInboxToken,
   });
@@ -143,12 +145,21 @@ export async function cancelOwnedTask(input: {
 
   // The task inbox may be closed after an earlier cancellation committed but
   // failed to stop its child. Retrying must still finish that cancellation.
-  await cancelTaskOwnedWork({
+  const forcedShutdown = await cancelTaskOwnedWork({
     cancelOwnedWork: input.cancelOwnedWork,
     entry: input.entry,
     serializedContext: input.serializedContext,
     session: input.session,
   });
+  if ((delivery === "unreachable" || forcedShutdown) && input.session !== undefined) {
+    // Forced shutdown can interrupt the lifecycle between its committed view
+    // and parent wake. Retried cancellation must finish delivery even when the
+    // inbox is gone; the shared delivery id deduplicates a wake already sent.
+    await wakeTaskParentStep({
+      token: sessionCommandHookToken(input.session.sessionId),
+      view,
+    });
+  }
   return view;
 }
 

packages/eve/src/execution/tasks/parent/task-cancel.ts

@@ -14,10 +14,10 @@ export type TaskExecutorCancel = (input: TaskExecutorCancelContext) => Promise<v
 const TASK_RUN_CANCEL_GRACE_MS = 1_000;
 const TASK_RUN_CANCEL_POLL_MS = 50;
 
-/** Cancels policy-specific work associated with one task. */
+/** Cancels task-owned work and reports whether the lifecycle run was forcibly stopped. */
 export async function cancelTaskOwnedWork(
   input: TaskExecutorCancelContext & { readonly cancelOwnedWork?: TaskExecutorCancel },
-): Promise<void> {
+): Promise<boolean> {
   const workflowToolRun = readWorkflowToolExecutorAddress(input.entry.executor);
   if (workflowToolRun !== undefined) {
     await cancelWorkflowToolRun(workflowToolRun, `Task ${input.entry.taskId} was cancelled.`);
@@ -27,9 +27,9 @@ export async function cancelTaskOwnedWork(
   while (Date.now() < deadline) {
     try {
       const status = await getRun(input.entry.taskRunId).status;
-      if (status !== "pending" && status !== "running") return;
+      if (status !== "pending" && status !== "running") return false;
     } catch {
-      return;
+      return false;
     }
     await new Promise((resolve) => setTimeout(resolve, TASK_RUN_CANCEL_POLL_MS));
   }
@@ -40,4 +40,5 @@ export async function cancelTaskOwnedWork(
   } catch {
     // The merged task run may have completed during its cooperative unwind.
   }
+  return true;
 }

packages/eve/src/harness/tool-loop.test.ts

@@ -1,3 +1,4 @@
+import { BoundaryHookError } from "#shared/boundary-hook-error.js";
 import { context as otelContext, trace } from "#compiled/@opentelemetry/api/index.js";
 import {
   type FilePart,
@@ -12858,3 +12859,124 @@ describe("appendMissingToolResultMessages", () => {
     ).toEqual([toolMessage]);
   });
 });
+
+describe("boundary event failures", () => {
+  it.each(["turn.started", "step.started"] as const)(
+    "parks a failed %s and accepts the next turn",
+    async (boundary) => {
+      const events: UnstampedMessageStreamEvent[] = [];
+      let denied = true;
+      const emit: HarnessEmitFn = async (event) => {
+        events.push(event);
+        if (denied && event.type === boundary)
+          throw new BoundaryHookError(new Error("admission denied"));
+      };
+      const runStep = createToolLoopHarness(createTestConfig("conversation", emit));
+      const result = await runStep(createTestSession({ outputSchema: { type: "object" } }), {
+        message: "Denied request",
+      });
+      expect(result.next).toBeNull();
+      expect(result.settledTurn).toEqual({ isError: true, output: "admission denied" });
+      expect(result.session.outputSchema).toBeUndefined();
+      expect(ToolLoopAgent).not.toHaveBeenCalled();
+      expect(events.filter((event) => event.type === "turn.failed")).toMatchObject([
+        { data: { turnId: "turn_0", sequence: 0, code: "EVENT_HANDLER_FAILED" } },
+      ]);
+      expect(events.map((event) => event.type)).toContain("session.waiting");
+      expect(events.map((event) => event.type)).not.toContain("session.failed");
+      expect(getHarnessEmissionState(result.session.state)).toEqual({
+        sessionStarted: true,
+        sequence: 1,
+        stepIndex: 0,
+        turnId: "",
+      });
+      denied = false;
+      setupMockAgent({
+        finishReason: "stop",
+        response: { messages: [{ role: "assistant", content: "recovered" }] },
+        text: "recovered",
+        toolCalls: [],
+        toolResults: [],
+      });
+      const recovered = await runStep(
+        JSON.parse(JSON.stringify(result.session)) as HarnessSession,
+        { message: "Try again" },
+      );
+      expect(recovered.next).toBeNull();
+      expect(events.filter((event) => event.type === "turn.completed")).toMatchObject([
+        { data: { turnId: "turn_1", sequence: 1 } },
+      ]);
+      expect(events.filter((event) => event.type === "session.started")).toHaveLength(1);
+    },
+  );
+
+  it("fails the current later step without invoking another model", async () => {
+    const events: UnstampedMessageStreamEvent[] = [];
+    const emit: HarnessEmitFn = async (event) => {
+      events.push(event);
+      if (event.type === "step.started")
+        throw new BoundaryHookError(new Error("step budget exhausted"));
+    };
+    const session = setHarnessEmissionState(createTestSession(), {
+      sessionStarted: true,
+      sequence: 2,
+      stepIndex: 3,
+      turnId: "turn_2",
+    });
+    const result = await createToolLoopHarness(createTestConfig("conversation", emit))(session);
+    expect(result.next).toBeNull();
+    expect(events[1]).toMatchObject({
+      type: "step.failed",
+      data: { stepIndex: 3, turnId: "turn_2" },
+    });
+    expect(ToolLoopAgent).not.toHaveBeenCalled();
+  });
+
+  it("lets a failed failure handler escalate", async () => {
+    const emit: HarnessEmitFn = async (event) => {
+      if (event.type === "turn.started") throw new BoundaryHookError(new Error("admission denied"));
+      if (event.type === "turn.failed") throw new Error("failure handler failed");
+    };
+    await expect(
+      createToolLoopHarness(createTestConfig("conversation", emit))(createTestSession(), {
+        message: "Hi",
+      }),
+    ).rejects.toThrow("failure handler failed");
+  });
+
+  it("keeps task failures terminal", async () => {
+    const events: UnstampedMessageStreamEvent[] = [];
+    const emit: HarnessEmitFn = async (event) => {
+      events.push(event);
+      if (event.type === "turn.started") throw new BoundaryHookError(new Error("task denied"));
+    };
+    await expect(
+      createToolLoopHarness(createTestConfig("task", emit))(createTestSession(), { message: "Hi" }),
+    ).rejects.toThrow("task denied");
+    expect(events.map((event) => event.type)).not.toContain("session.waiting");
+  });
+
+  it("keeps runtime preamble failures terminal", async () => {
+    const failure = new Error("memory recall failed");
+    const emit: HarnessEmitFn = async (event) => {
+      if (event.type === "turn.started") throw failure;
+    };
+    await expect(
+      createToolLoopHarness(createTestConfig("conversation", emit))(createTestSession(), {
+        message: "Hi",
+      }),
+    ).rejects.toBe(failure);
+  });
+
+  it("preserves explicit cancellation from a boundary handler", async () => {
+    const cancellation = new TurnCancelledError();
+    const emit: HarnessEmitFn = async () => {
+      throw cancellation;
+    };
+    await expect(
+      createToolLoopHarness(createTestConfig("conversation", emit))(createTestSession(), {
+        message: "Hi",
+      }),
+    ).rejects.toBe(cancellation);
+  });
+});

packages/eve/src/harness/tool-loop.ts

@@ -1,3 +1,4 @@
+import { BoundaryHookError } from "#shared/boundary-hook-error.js";
 import {
   isStepCount,
   type LanguageModelCallEndEvent,
@@ -189,7 +190,7 @@ import {
   extractUpstreamRejectionMessage,
 } from "#harness/model-call-error.js";
 import { summarizeKnownError, type SemanticErrorSummary } from "#harness/semantic-errors/index.js";
-import { throwIfTurnAborted } from "#harness/turn-cancellation.js";
+import { isTurnCancellation, throwIfTurnAborted } from "#harness/turn-cancellation.js";
 import type { JsonObject, JsonValue } from "#shared/json.js";
 import { EMPTY_DELIVERY_SENTINEL, hasEmptyDeliverySentinel } from "#shared/empty-delivery.js";
 import { resolveDeliveryPolicy } from "#tasks/delivery-policy.js";
@@ -537,6 +538,38 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
         session,
       };
     };
+    const failBoundaryEvent = async (
+      error: unknown,
+      failureState: ReturnType<typeof getHarnessEmissionState>,
+    ): Promise<StepResult> => {
+      throwIfTurnAborted(config.abortSignal);
+      if (isTurnCancellation(error)) throw error;
+      if (isDynamicModelSelectionError(error)) return failModelSelection(error, failureState);
+      if (!emit || config.mode !== "conversation" || !(error instanceof BoundaryHookError)) {
+        throw error;
+      }
+
+      stepInstrumentation?.recordError(error);
+      const errorId = createErrorId();
+      const message = toErrorMessage(error);
+      log.error("turn boundary handler failed — parking session", {
+        error,
+        errorId,
+        sessionId: session.sessionId,
+        turnId: failureState.turnId,
+      });
+      emissionState = await emitRecoverableFailedTurn(emit, failureState, {
+        code: "EVENT_HANDLER_FAILED",
+        continuationToken: session.continuationToken,
+        details: { errorId },
+        message,
+      });
+      return {
+        next: null,
+        session: setHarnessEmissionState({ ...session, outputSchema: undefined }, emissionState),
+        settledTurn: { isError: true, output: message },
+      };
+    };
     const preparePreambleTrace = async (): Promise<RuntimeTraceContext | undefined> => {
       return await stepInstrumentation?.preparePreamble({
         sequence: emissionState.sequence,
@@ -856,8 +889,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
             history: [...parkedSession.history, ...instructionMessages],
           };
           session = parkedSession;
-          if (!isDynamicModelSelectionError(error)) throw error;
-          return failModelSelection(error, {
+          return failBoundaryEvent(error, {
             sessionStarted: true,
             sequence: emissionState.sequence,
             stepIndex: 0,
@@ -989,8 +1021,7 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
           history: [...(memoryCommit?.history ?? pending.session.history), ...instructionMessages],
           state: memoryCommit?.state ?? pending.session.state,
         };
-        if (!isDynamicModelSelectionError(error)) throw error;
-        return failModelSelection(error, {
+        return failBoundaryEvent(error, {
           sessionStarted: true,
           sequence: emissionState.sequence,
           stepIndex: 0,
@@ -1175,12 +1206,16 @@ export function createToolLoopHarness(config: ToolLoopHarnessConfig): StepFn {
     );
 
     if (emit) {
-      await emitStepStarted(
-        emit,
-        emissionState,
-        requireSessionModelReference(session).id,
-        projectedMessages,
-      );
+      try {
+        await emitStepStarted(
+          emit,
+          emissionState,
+          requireSessionModelReference(session).id,
+          projectedMessages,
+        );
+      } catch (error) {
+        return failBoundaryEvent(error, emissionState);
+      }
     }
     const approvedTools = getApprovedTools(
       session,

packages/eve/src/internal/testing/task-cancel-notification-workflow.ts

@@ -0,0 +1,77 @@
+import { createHook, getWorkflowMetadata, sleep } from "#compiled/@workflow/core/index.js";
+
+import { createSessionCommandInbox } from "#execution/session-command-inbox.js";
+import { sessionCommandHookToken } from "#execution/session-command-token.js";
+import { appendTaskViewStep } from "#execution/tasks/child/steps.js";
+import { cancelOwnedTask } from "#execution/tasks/parent/dispatch.js";
+import { waitForCommandHookOwner } from "#execution/workflow-runtime.js";
+import { getRun, start } from "#internal/workflow/runtime.js";
+import type { HarnessSession } from "#harness/types.js";
+import type { SessionTaskIndexEntry } from "#tasks/session-index.js";
+import type { TaskCommandHookPayload } from "#tasks/types.js";
+
+/** Models a task whose view commits before its executor finishes unwinding. */
+export async function slowCancelledTaskWorkflow(input: {
+  readonly taskId: string;
+  readonly taskInboxToken: string;
+}): Promise<void> {
+  "use workflow";
+
+  using commands = createHook<TaskCommandHookPayload>({ token: input.taskInboxToken });
+  const metadata = { kind: "tool", name: "slow-cancel" } as const;
+  await appendTaskViewStep({ view: { metadata, status: "working", taskId: input.taskId } });
+  const delivery = await commands;
+  if (delivery.kind !== "task-command" || delivery.command.kind !== "cancel") {
+    throw new Error("Expected the task cancellation command.");
+  }
+  await appendTaskViewStep({ view: { metadata, status: "cancelled", taskId: input.taskId } });
+  await sleep("1h");
+}
+
+export async function startSlowCancelledTaskStep(input: {
+  readonly sessionId: string;
+}): Promise<SessionTaskIndexEntry> {
+  "use step";
+
+  const taskId = `${input.sessionId}-task`;
+  const taskInboxToken = `${input.sessionId}:slow-cancel`;
+  const run = await start(slowCancelledTaskWorkflow, [{ taskId, taskInboxToken }]);
+  await waitForCommandHookOwner(taskInboxToken);
+  return {
+    createdByTurnId: "turn_0",
+    metadata: { kind: "tool", name: "slow-cancel" },
+    taskId,
+    taskInboxToken,
+    taskRunId: run.runId,
+  };
+}
+
+export async function cancelSlowTaskFromParentStep(input: {
+  readonly entry: SessionTaskIndexEntry;
+  readonly sessionId: string;
+}) {
+  "use step";
+
+  const view = await cancelOwnedTask({
+    entry: input.entry,
+    session: { sessionId: input.sessionId } as HarnessSession,
+  });
+  return { view, taskRunStatus: await getRun(input.entry.taskRunId).status };
+}
+
+export async function taskCancelNotificationWorkflow() {
+  "use workflow";
+
+  const { workflowRunId: sessionId } = getWorkflowMetadata();
+  const inbox = createSessionCommandInbox();
+  try {
+    await inbox.claimStable(sessionCommandHookToken(sessionId));
+    const entry = await startSlowCancelledTaskStep({ sessionId });
+    const cancelled = await cancelSlowTaskFromParentStep({ entry, sessionId });
+    const next = await inbox.next();
+    inbox.consumeNext();
+    return { ...cancelled, notification: next.value };
+  } finally {
+    await inbox.dispose();
+  }
+}

packages/eve/src/shared/boundary-hook-error.ts

@@ -0,0 +1,9 @@
+import { toErrorMessage } from "#shared/errors.js";
+
+/** Distinguishes authored admission failures from failed runtime lifecycle work. */
+export class BoundaryHookError extends Error {
+  constructor(cause: unknown) {
+    super(toErrorMessage(cause), { cause });
+    this.name = "BoundaryHookError";
+  }
+}