diff --git a/packages/junior/src/chat/agent-invocations/AGENTS.md b/packages/junior/src/chat/agent-invocations/AGENTS.md
index 74b989adc4..df29641179 100644
--- a/packages/junior/src/chat/agent-invocations/AGENTS.md
+++ b/packages/junior/src/chat/agent-invocations/AGENTS.md
@@ -31,6 +31,10 @@ reads. Terms are in `TERMINOLOGY.md`.
   no boundary is left, fail it now. Do not throw and requeue forever.
 - The parent gets one report for the child's work. A finished invocation is
   `replaced` while the child still has open work.
+- Pick the child worker from the stored `parentConversationId`. Never pick it
+  from the Turn Source: a child watch-event Turn has an event Source. All
+  child work goes to the child worker and never falls through to a root
+  worker.
 
 ## Authority and privacy
 
diff --git a/packages/junior/src/chat/agent-invocations/result.ts b/packages/junior/src/chat/agent-invocations/result.ts
index 583d0e9268..5253baee50 100644
--- a/packages/junior/src/chat/agent-invocations/result.ts
+++ b/packages/junior/src/chat/agent-invocations/result.ts
@@ -2,6 +2,7 @@ import type { StateAdapter } from "chat";
 import type { Actor } from "@/chat/actor";
 import { credentialContextForActor } from "@/chat/credentials/context";
 import { SYSTEM_IDENTITY_ID } from "@/chat/identities/actor";
+import type { AgentRunner } from "@/chat/runtime/agent-runner";
 import {
   EVENT_WAIT_MS,
   isEventMailboxMetadata,
@@ -40,7 +41,11 @@ import type {
   ChildAgent,
   TerminalAgentInvocation,
 } from "./types";
-import { enqueueAgentInvocation } from "./work";
+import {
+  createAgentInvocationWorker,
+  enqueueAgentInvocation,
+  resolveAgentInvocationId,
+} from "./work";
 
 const AGENT_INVOCATION_EVENT_NAMESPACE = "junior";
 // TODO(dcramer): Add an `agent_invocation.needs_input` Event type when a child
@@ -360,7 +365,7 @@ async function getChildAgentInputInvocation(
  * reports to the child's parent. Returns undefined for work that is not
  * mailbox input for a child Conversation.
  */
-export async function startAgentInvocationsForChildInput(
+async function startAgentInvocationsForChildInput(
   context: ConversationWorkerContext,
   options: { queue: ConversationWorkQueue; state?: StateAdapter },
 ): Promise<ConversationWorkerResult | undefined> {
@@ -443,6 +448,39 @@ export async function startAgentInvocationsForChildInput(
   return { status: "completed" };
 }
 
+/**
+ * Build the worker for all work of a child Conversation. The caller selects
+ * it from the stored parent of the Conversation, never from a Turn Source. It
+ * starts agent invocations for child input, then runs the agent invocation
+ * that owns the work. Work that no agent invocation owns is an error.
+ */
+export function createChildConversationWorker(
+  agentRunner: AgentRunner,
+  options: {
+    enqueueResult?: (invocation: TerminalAgentInvocation) => Promise<void>;
+    maxAgentDepth?: number;
+    queue: ConversationWorkQueue;
+    state?: StateAdapter;
+  },
+) {
+  const invocationWorker = createAgentInvocationWorker(agentRunner, options);
+  return async (
+    context: ConversationWorkerContext,
+  ): Promise<ConversationWorkerResult> => {
+    const started = await startAgentInvocationsForChildInput(context, options);
+    if (started) {
+      return started;
+    }
+    const invocationId = await resolveAgentInvocationId(context);
+    if (!invocationId) {
+      throw new Error(
+        `Child conversation ${context.conversationId} has no agent invocation for this work`,
+      );
+    }
+    return await invocationWorker(context, invocationId);
+  };
+}
+
 /**
  * Run each agent invocation result or message as the Actor that triggered the
  * child, user or system. This is on purpose: the result continues that
diff --git a/packages/junior/src/chat/agent-invocations/work.ts b/packages/junior/src/chat/agent-invocations/work.ts
index 5c7f8a4416..ba29fb96e9 100644
--- a/packages/junior/src/chat/agent-invocations/work.ts
+++ b/packages/junior/src/chat/agent-invocations/work.ts
@@ -257,12 +257,15 @@ export async function resolveAgentInvocationId(
   if (active) {
     return active.invocationId;
   }
-  // A stop request can outlive the run that it stopped and wake the child
-  // again. The stopped agent invocation handles that wake.
+  // A child Conversation never runs on a root worker. A wake after its last
+  // agent invocation ended, such as a late stop request, goes to that agent
+  // invocation, which only repeats its idempotent finish. A paused Turn of an
+  // ended agent invocation can have an event Source, and a root worker would
+  // try to resume it.
   const latest = await getLatestAgentInvocationForConversation(
     context.conversationId,
   );
-  return latest?.status === "stopped" ? latest.invocationId : undefined;
+  return latest?.invocationId;
 }
 
 /**
@@ -396,12 +399,15 @@ async function persistTerminalLifecycle(
   terminal: TerminalAgentInvocation,
 ): Promise<void> {
   const turnId = getAgentInvocationTurnId(terminal.invocationId);
-  if (terminal.status === "stopped") {
-    // A child stopped between slices has a paused Turn record. Nothing may
+  if (terminal.status !== "completed") {
+    // A child that ends between slices has a paused Turn record. Nothing may
     // resume it.
     await abandonTurnRecord({
       conversationId: terminal.childConversationId,
-      errorMessage: "Turn cancelled",
+      errorMessage:
+        terminal.status === "stopped"
+          ? "Turn cancelled"
+          : terminal.errorMessage,
       turnId,
     });
   }
@@ -787,10 +793,9 @@ export function createAgentInvocationWorker(
           }),
       );
     } catch (error) {
-      if (!(error instanceof AgentRunError)) {
-        throw error;
-      }
-      const runError = error.cause;
+      // A save error has no AgentRunError wrapper. On the final attempt it
+      // must also fail the agent invocation, or the child stays open.
+      const runError = error instanceof AgentRunError ? error.cause : error;
       if (isInvocationInputCommitLost(runError)) {
         return { status: "lost_lease" };
       }
@@ -869,23 +874,3 @@ export function createAgentInvocationWorker(
     return { status: "completed" };
   };
 }
-
-/** Route invocation-backed work before falling through to existing consumers. */
-export function routeAgentInvocationWork(options: {
-  fallbackWorker: (
-    context: ConversationWorkerContext,
-  ) => Promise<ConversationWorkerResult>;
-  invocationWorker: (
-    context: ConversationWorkerContext,
-    invocationId: string,
-  ) => Promise<ConversationWorkerResult>;
-}) {
-  return async (
-    context: ConversationWorkerContext,
-  ): Promise<ConversationWorkerResult> => {
-    const invocationId = await resolveAgentInvocationId(context);
-    return invocationId
-      ? await options.invocationWorker(context, invocationId)
-      : await options.fallbackWorker(context);
-  };
-}
diff --git a/packages/junior/src/chat/app/conversation-work.ts b/packages/junior/src/chat/app/conversation-work.ts
index 4052fa079b..fa6986f3d6 100644
--- a/packages/junior/src/chat/app/conversation-work.ts
+++ b/packages/junior/src/chat/app/conversation-work.ts
@@ -17,12 +17,8 @@ import {
   resolveAgentDispatchId,
 } from "@/chat/agent-dispatch/work";
 import {
-  createAgentInvocationWorker,
-  resolveAgentInvocationId,
-} from "@/chat/agent-invocations/work";
-import {
+  createChildConversationWorker,
   enqueueAgentInvocationResult,
-  startAgentInvocationsForChildInput,
   withAgentInvocationResultActors,
 } from "@/chat/agent-invocations/result";
 import { botConfig } from "@/chat/config";
@@ -143,13 +139,15 @@ export function createConversationWork(
     },
     runTurn: runtime.runDispatchTurn,
   });
-  const invocationWorker = createAgentInvocationWorker(options.agentRunner, {
+  const childWorker = createChildConversationWorker(options.agentRunner, {
     enqueueResult: async (invocation) =>
       await enqueueAgentInvocationResult(invocation, {
         queue: options.queue,
         state: options.state,
       }),
     maxAgentDepth: botConfig.maxAgentDepth,
+    queue: options.queue,
+    state: options.state,
   });
   const conversationTurnWorker = createConversationTurnWorker(
     options.agentRunner,
@@ -158,8 +156,8 @@ export function createConversationWork(
       state: options.state,
     }),
   );
-  // A Slack Destination uses the Slack worker. All other work must have been
-  // selected above this function.
+  // A Slack Destination uses the Slack worker. All other root work must have
+  // been selected above this function.
   const destinationWorker = async (
     context: ConversationWorkerContext,
   ): Promise<ConversationWorkerResult> => {
@@ -184,15 +182,13 @@ export function createConversationWork(
     if (skipped) {
       return skipped;
     }
-    // Mailbox input for a child Conversation (a result, a message, or a watch
-    // event) starts an agent invocation on that child. The same input for a
-    // root Conversation runs as an Event Turn below.
-    const childInput = await startAgentInvocationsForChildInput(context, {
-      queue: options.queue,
-      state: options.state,
+    // The stored parent decides the kind of Conversation, so child work never
+    // reaches a root worker. A child Turn can have any Source.
+    const conversation = await options.conversationStore.get({
+      conversationId: context.conversationId,
     });
-    if (childInput) {
-      return childInput;
+    if (conversation?.parentConversationId) {
+      return await childWorker(context);
     }
     const mailboxTurn = await resolveMailboxTurnWork(context);
     if (mailboxTurn) {
@@ -201,10 +197,6 @@ export function createConversationWork(
         await withAgentInvocationResultActors(mailboxTurn),
       );
     }
-    const invocationId = await resolveAgentInvocationId(context);
-    if (invocationId) {
-      return await invocationWorker(context, invocationId);
-    }
     const dispatchId = await resolveAgentDispatchId(context);
     if (dispatchId) {
       return await dispatchWorker(context, dispatchId);
diff --git a/packages/junior/src/chat/task-execution/AGENTS.md b/packages/junior/src/chat/task-execution/AGENTS.md
index 2446d684eb..2cbb032533 100644
--- a/packages/junior/src/chat/task-execution/AGENTS.md
+++ b/packages/junior/src/chat/task-execution/AGENTS.md
@@ -53,6 +53,9 @@ checkpoint, paused Turns, heartbeat recovery, and the shared worker loop in
   new queue wake.
 - A Turn that loses its worker resumes once from its last save. A second loss
   fails the Turn and keeps committed history.
+- The last attempt before the worker gives up is final, also for a resume
+  wake with no messages. On that attempt each worker records its own
+  failure, so no work stops silently.
 - Lease expiry permits recovery. It must not erase mailbox work or agent
   history.
 - Heartbeats repair missing wakes and abandoned leases. They are not a second
diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts
index 4c472587e5..5934121602 100644
--- a/packages/junior/src/chat/task-execution/worker.ts
+++ b/packages/junior/src/chat/task-execution/worker.ts
@@ -71,6 +71,11 @@ export interface InboxAttempt {
     handle: (messages: InboundMessage[]) => Promise<readonly string[] | void>,
     options?: { deferred?: boolean },
   ): Promise<InboundMessage[]>;
+  /**
+   * True when a failure of this attempt can end the work: the worker gives up
+   * and drops the mailbox. The adapter must record its own failure first.
+   * Resume attempts, which have no messages, can also be final.
+   */
   isFinalAttempt: boolean;
   messages: InboundMessage[];
 }
@@ -226,6 +231,21 @@ function isTerminalFailure(failure: AttemptFailure): boolean {
   );
 }
 
+/**
+ * Return whether a failure of this attempt can end the conversation work. The
+ * retry limit applies to every attempt, also one with no messages, such as a
+ * resume. A message also ends it at its own delivery limit.
+ */
+function isLastConversationAttempt(
+  work: ConversationWorkState,
+  messages: readonly InboundMessage[],
+): boolean {
+  return (
+    (work.execution.retryCount ?? 0) + 1 >= CONVERSATION_WORK_MAX_RETRIES ||
+    messages.some((message) => isFinalAttempt(message))
+  );
+}
+
 function logRetryExhausted(args: {
   error: unknown;
   retryCount: number;
@@ -636,8 +656,9 @@ async function processConversationWorkInContext(
           conversationId,
           destination,
           drain,
-          isFinalAttempt: attemptMessages.some((message) =>
-            isFinalAttempt(message),
+          isFinalAttempt: isLastConversationAttempt(
+            leasedWork,
+            attemptMessages,
           ),
           messages: attemptMessages,
         },
diff --git a/packages/junior/src/cli/chat.ts b/packages/junior/src/cli/chat.ts
index 6207bb5f72..906c85e878 100644
--- a/packages/junior/src/cli/chat.ts
+++ b/packages/junior/src/cli/chat.ts
@@ -246,9 +246,7 @@ async function prepareLocalChatRun(
     await import("@/chat/local/sandbox-egress-signals");
   const { bindSpawnAgent } = await import("@/chat/agent-invocations/spawn");
   const { bindMessageAgent } = await import("@/chat/agent-invocations/message");
-  const { createAgentInvocationWorker, routeAgentInvocationWork } =
-    await import("@/chat/agent-invocations/work");
-  const { startAgentInvocationsForChildInput } =
+  const { createChildConversationWorker } =
     await import("@/chat/agent-invocations/result");
   const { skipInputBeforeConversationStop } =
     await import("@/chat/conversations/session-stop");
@@ -261,26 +259,15 @@ async function prepareLocalChatRun(
     if (!agentRunner) {
       throw new Error("Local agent runner is not ready");
     }
-    const run = routeAgentInvocationWork({
-      // The local queue carries only child Conversation work.
-      fallbackWorker: async (context) => {
-        const skipped = await skipInputBeforeConversationStop(context);
-        if (skipped) {
-          return skipped;
-        }
-        const started = await startAgentInvocationsForChildInput(context, {
-          queue: localConversationWork.queue,
-        });
-        if (!started) {
-          throw new Error("Local child queue received non-invocation work");
-        }
-        return started;
-      },
-      invocationWorker: createAgentInvocationWorker(agentRunner),
+    const childWorker = createChildConversationWorker(agentRunner, {
+      queue: localConversationWork.queue,
     });
     await processConversationWork(message, {
       queue: localConversationWork.queue,
-      run,
+      // The local queue carries only child Conversation work.
+      run: async (context) =>
+        (await skipInputBeforeConversationStop(context)) ??
+        (await childWorker(context)),
     });
   });
   agentRunner = createAgentRunner(executeAgentRun, {
diff --git a/packages/junior/tests/component/agent-invocation-worker.test.ts b/packages/junior/tests/component/agent-invocation-worker.test.ts
index 22513c8ca2..0b491dafc7 100644
--- a/packages/junior/tests/component/agent-invocation-worker.test.ts
+++ b/packages/junior/tests/component/agent-invocation-worker.test.ts
@@ -1,7 +1,7 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { PiMessage } from "@/chat/pi/messages";
 import { buildSteeringPiMessage } from "@/chat/agent/prompt";
-import type { AgentSteeringMessage } from "@/chat/agent/types";
+import type { AgentRun, AgentSteeringMessage } from "@/chat/agent/types";
 import { agentInstructionProvenance } from "@/chat/conversations/provenance";
 import {
   createAgentInvocation,
@@ -523,6 +523,57 @@ describe("agent invocation worker", () => {
     }
   });
 
+  it("lets child agents spawn only down to maxAgentDepth", async () => {
+    const fixture = await prepareParentConversation();
+    try {
+      const child = await createAgentInvocation({
+        ...INVOCATION_INPUT,
+        idempotencyKey: "depth-child",
+      });
+      const helper = await createAgentInvocation({
+        ...INVOCATION_INPUT,
+        idempotencyKey: "depth-helper",
+        parentConversationId: child.childConversationId,
+      });
+      const runs: AgentRun[] = [];
+      const worker = createAgentInvocationWorker(
+        {
+          run: async (run) => {
+            runs.push(run);
+            throw new Error("stop after reading the run");
+          },
+        },
+        { maxAgentDepth: 1 },
+      );
+      // With maxAgentDepth 1, depth 1 can spawn and depth 2 cannot.
+      for (const invocation of [child, helper]) {
+        await expect(
+          worker(
+            {
+              attempt: {
+                ack: vi.fn(),
+                conversationId: invocation.childConversationId,
+                drain: vi.fn(),
+                isFinalAttempt: false,
+                messages: [buildAgentInvocationInboundMessage(invocation)],
+              },
+              checkIn: vi.fn(),
+              conversationId: invocation.childConversationId,
+              shouldYield: () => false,
+            },
+            invocation.invocationId,
+          ),
+        ).rejects.toThrow("stop after reading the run");
+      }
+      expect(runs.map((run) => run.disabledFeatures)).toEqual([
+        ["handoff", "interactive-auth"],
+        ["handoff", "interactive-auth", "subagents"],
+      ]);
+    } finally {
+      await fixture.close();
+    }
+  });
+
   it("keeps unexpected runner failures retryable until the final attempt", async () => {
     const fixture = await prepareParentConversation();
     try {
diff --git a/packages/junior/tests/fixtures/agent-dispatch.ts b/packages/junior/tests/fixtures/agent-dispatch.ts
index a15df79766..5c4b6aea97 100644
--- a/packages/junior/tests/fixtures/agent-dispatch.ts
+++ b/packages/junior/tests/fixtures/agent-dispatch.ts
@@ -5,13 +5,7 @@ import type {
 } from "@sentry/junior-plugin-api";
 import { createOrGetDispatch } from "@/chat/agent-dispatch/store";
 import type { BoundDispatchOptions } from "@/chat/agent-dispatch/types";
-import { createConversationWork } from "@/chat/app/conversation-work";
-import { createJuniorSlackAdapter } from "@/chat/slack/adapter";
 import type { CredentialSubject } from "@/chat/credentials/context";
-import type { AgentRunner } from "@/chat/runtime/agent-runner";
-import { getConversationStore } from "@/chat/db";
-import { getStateAdapter } from "@/chat/state/adapter";
-import { createConversationWorkQueueTestAdapter } from "./conversation-work";
 
 export const agentDispatchTestDestination = {
   platform: "slack",
@@ -19,26 +13,6 @@ export const agentDispatchTestDestination = {
   channelId: "C123",
 } as const;
 
-/** Compose production conversation work for one agent dispatch test. */
-export async function createAgentDispatchWorkHarness(agentRunner: AgentRunner) {
-  const state = getStateAdapter();
-  await state.connect();
-  const queue = createConversationWorkQueueTestAdapter();
-  const adapter = createJuniorSlackAdapter({
-    botToken: "xoxb-test",
-    botUserId: "U0BOT",
-    signingSecret: "test-signing-secret",
-  });
-  const work = createConversationWork({
-    agentRunner,
-    conversationStore: getConversationStore(),
-    getSlackAdapter: () => adapter,
-    queue,
-    state,
-  });
-  return { queue, run: work.run, runtime: work.runtime, state };
-}
-
 /** Create a durable dispatch record for integration tests. */
 export async function createAgentDispatchTestRecord(
   idempotencyKey: string,
diff --git a/packages/junior/tests/fixtures/conversation-work.ts b/packages/junior/tests/fixtures/conversation-work.ts
index 2e00e975fa..b4cd04b75a 100644
--- a/packages/junior/tests/fixtures/conversation-work.ts
+++ b/packages/junior/tests/fixtures/conversation-work.ts
@@ -9,6 +9,7 @@ import {
 } from "@earendil-works/pi-ai/providers/faux";
 import type { Destination } from "@sentry/junior-plugin-api";
 import { createConversationWork } from "@/chat/app/conversation-work";
+import type { JuniorRuntimeServiceOverrides } from "@/chat/app/services";
 import { executeAgentRun } from "@/chat/agent";
 import { getConversationStore } from "@/chat/db";
 import type { AgentRunner } from "@/chat/runtime/agent-runner";
@@ -154,6 +155,19 @@ export class ConversationWorkQueueTestAdapter implements ConversationWorkQueue {
     return entry.message;
   }
 
+  /** Deliver the next payload for one Conversation and keep the others queued. */
+  takeMessageFor(conversationId: string): ConversationQueueMessage {
+    const index = this.#queuedMessages.findIndex(
+      (entry) => entry.message.conversationId === conversationId,
+    );
+    if (index === -1) {
+      throw new Error(
+        `Expected queued conversation work for ${conversationId}`,
+      );
+    }
+    return this.#queuedMessages.splice(index, 1)[0]!.message;
+  }
+
   /** Honor queue delays when running evals against the real clock. */
   async takeReadyMessage(
     signal?: AbortSignal,
@@ -435,6 +449,48 @@ export type ConversationWorkSlackHarness = {
   }) => Promise<Response>;
 };
 
+/**
+ * Compose production conversation work once for one test. `process` runs one
+ * queue wake through the production worker, as the Vercel queue callback does.
+ * Pass `services` only to fake a model boundary that the work does not own.
+ */
+export async function createConversationWorkHarness(
+  agentRunner: AgentRunner,
+  options: { services?: JuniorRuntimeServiceOverrides } = {},
+) {
+  const state = getStateAdapter();
+  await state.connect();
+  const queue = createConversationWorkQueueTestAdapter();
+  const adapter = createSlackAdapterFixture();
+  const conversationStore = getConversationStore();
+  const work = createConversationWork({
+    agentRunner,
+    conversationStore,
+    getSlackAdapter: () => adapter,
+    queue,
+    ...(options.services ? { services: options.services } : undefined),
+    state,
+  });
+  return {
+    adapter,
+    process: async (
+      message: ConversationQueueMessage,
+      processOptions: { softYieldAfterMs?: number } = {},
+    ) =>
+      await processConversationQueueMessage(message, {
+        conversationStore,
+        queue,
+        run: work.run,
+        state,
+        ...processOptions,
+      }),
+    queue,
+    run: work.run,
+    runtime: work.runtime,
+    state,
+  };
+}
+
 /**
  * Compose the production Slack ingress → durable queue → worker → agent path.
  * Fake only model generation at the executeAgentRun stream boundary and Slack
@@ -446,19 +502,11 @@ export async function createConversationWorkSlackHarness(
     modelStream?: StreamFn;
   } = {},
 ): Promise<ConversationWorkSlackHarness> {
-  const state = getStateAdapter();
-  await state.connect();
-  const wakes = createConversationWorkQueueTestAdapter();
-  const adapter = createSlackAdapterFixture();
   const modelStream = options.modelStream ?? streamReplies("Deploy checked.");
   const agentRunner: AgentRunner = options.agentRunner ?? {
     run: async (request) => await executeAgentRun(request, modelStream),
   };
-  const work = createConversationWork({
-    agentRunner,
-    conversationStore: getConversationStore(),
-    getSlackAdapter: () => adapter,
-    queue: wakes,
+  const work = await createConversationWorkHarness(agentRunner, {
     services: {
       agentRunner,
       subscribedReplyPolicy: {
@@ -470,8 +518,9 @@ export async function createConversationWorkSlackHarness(
         listThreadReplies: async () => [],
       },
     },
-    state,
   });
+  const { adapter, state } = work;
+  const wakes = work.queue;
 
   const services = {
     getSlackAdapter: () => adapter,
@@ -490,23 +539,14 @@ export async function createConversationWorkSlackHarness(
         .map((call) => call.params.text)
         .filter((text): text is string => typeof text === "string"),
     next: async (requestStartedAtMs?: number) => {
-      const process = async () =>
-        await processConversationQueueMessage(wakes.takeMessage(), {
-          queue: wakes,
-          run: work.run,
-          state,
-        });
+      const process = async () => await work.process(wakes.takeMessage());
       return requestStartedAtMs === undefined
         ? await process()
         : await runWithTurnRequestDeadline(process, requestStartedAtMs);
     },
     drain: async () => {
       for (let i = 0; i < 12 && wakes.hasQueuedMessages(); i += 1) {
-        await processConversationQueueMessage(wakes.takeMessage(), {
-          queue: wakes,
-          run: work.run,
-          state,
-        });
+        await work.process(wakes.takeMessage());
       }
       if (wakes.hasQueuedMessages()) {
         throw new Error("queue still has work after drain");
diff --git a/packages/junior/tests/integration/agent-dispatch-recovery.test.ts b/packages/junior/tests/integration/agent-dispatch-recovery.test.ts
index 11b6076a07..402723061f 100644
--- a/packages/junior/tests/integration/agent-dispatch-recovery.test.ts
+++ b/packages/junior/tests/integration/agent-dispatch-recovery.test.ts
@@ -8,7 +8,10 @@ import {
 import { recoverPendingDispatchMailboxAppends } from "@/chat/agent-dispatch/heartbeat";
 import { enqueueAgentDispatch } from "@/chat/agent-dispatch/work";
 import { disconnectStateAdapter } from "@/chat/state/adapter";
-import { createConversationWorkQueueTestAdapter } from "../fixtures/conversation-work";
+import {
+  createConversationWorkHarness,
+  createConversationWorkQueueTestAdapter,
+} from "../fixtures/conversation-work";
 import { processConversationQueueMessage } from "@/chat/task-execution/vercel-callback";
 import { getTurnRecord } from "@/chat/task-execution/turn-cursor";
 import { slackApiOutbox } from "../fixtures/slack-api-outbox";
@@ -21,7 +24,6 @@ import { createModelStream } from "../fixtures/model-stream";
 import {
   agentDispatchTestDestination as destination,
   createAgentDispatchTestRecord as createDispatch,
-  createAgentDispatchWorkHarness,
 } from "../fixtures/agent-dispatch";
 
 vi.hoisted(() => {
@@ -41,7 +43,7 @@ describe("agent dispatch recovery", () => {
 
   it("projects the diagnostic from a terminal model failure", async () => {
     const dispatch = await createDispatch("model-failure-detail");
-    const firstRun = await createAgentDispatchWorkHarness(
+    const firstRun = await createConversationWorkHarness(
       createModelAgentRunner(
         createModelStream([
           {
@@ -74,7 +76,7 @@ describe("agent dispatch recovery", () => {
       errorMessage: "Model provider quota exhausted",
     });
 
-    const replay = await createAgentDispatchWorkHarness(neverRunAgentRunner());
+    const replay = await createConversationWorkHarness(neverRunAgentRunner());
     await enqueueAgentDispatch(dispatch, {
       queue: replay.queue,
       state: replay.state,
@@ -142,7 +144,7 @@ describe("agent dispatch recovery", () => {
     );
     const runAgent = vi.spyOn(agentRunner, "run");
     const { queue, run, state } =
-      await createAgentDispatchWorkHarness(agentRunner);
+      await createConversationWorkHarness(agentRunner);
 
     await enqueueAgentDispatch(dispatch, { queue, state });
     let deliveries = 0;
diff --git a/packages/junior/tests/integration/agent-dispatch-work.test.ts b/packages/junior/tests/integration/agent-dispatch-work.test.ts
index bcd699d3f4..ac6dec8ea2 100644
--- a/packages/junior/tests/integration/agent-dispatch-work.test.ts
+++ b/packages/junior/tests/integration/agent-dispatch-work.test.ts
@@ -31,8 +31,8 @@ import {
 import {
   agentDispatchTestDestination as destination,
   createAgentDispatchTestRecord as createDispatch,
-  createAgentDispatchWorkHarness,
 } from "../fixtures/agent-dispatch";
+import { createConversationWorkHarness } from "../fixtures/conversation-work";
 
 vi.hoisted(() => {
   process.env.JUNIOR_STATE_ADAPTER = "memory";
@@ -70,7 +70,7 @@ describe("agent dispatch conversation work", () => {
       input,
     );
     const modelStream = vi.fn(createModelStream([sendMessage("Done")]));
-    const { queue, run, state } = await createAgentDispatchWorkHarness(
+    const { queue, run, state } = await createConversationWorkHarness(
       createModelAgentRunner(modelStream),
     );
 
@@ -148,7 +148,7 @@ describe("agent dispatch conversation work", () => {
       queue,
       run: runWork,
       state,
-    } = await createAgentDispatchWorkHarness(agentRunner);
+    } = await createConversationWorkHarness(agentRunner);
 
     const titleRequests: TitleModelRequest[] = [];
     mockTitleModel("Weekly digest for the platform team", {
@@ -237,7 +237,7 @@ describe("agent dispatch conversation work", () => {
       ],
       { ...destination, threadTs: "1700000000.000200" },
     );
-    const { queue, run, state } = await createAgentDispatchWorkHarness(
+    const { queue, run, state } = await createConversationWorkHarness(
       createModelAgentRunner(
         createModelStream([sendMessage("Scheduled digest")]),
       ),
@@ -275,7 +275,7 @@ describe("agent dispatch conversation work", () => {
         },
       ],
     );
-    const { queue, run, state } = await createAgentDispatchWorkHarness(
+    const { queue, run, state } = await createConversationWorkHarness(
       createModelAgentRunner(createModelStream([sendMessage("Work complete")])),
     );
 
@@ -320,7 +320,7 @@ describe("agent dispatch conversation work", () => {
     );
     const runAgent = vi.spyOn(agentRunner, "run");
     const { queue, run, state } =
-      await createAgentDispatchWorkHarness(agentRunner);
+      await createConversationWorkHarness(agentRunner);
     const processNext = async (softYieldAfterMs?: number) =>
       await processConversationQueueMessage(queue.takeMessage(), {
         queue,
@@ -396,7 +396,7 @@ describe("agent dispatch conversation work", () => {
       },
     );
     await persistConversationMessages({ conversation, conversationId });
-    const { queue, run, state } = await createAgentDispatchWorkHarness(
+    const { queue, run, state } = await createConversationWorkHarness(
       neverRunAgentRunner(),
     );
 
diff --git a/packages/junior/tests/integration/agent-invocation-concurrency.test.ts b/packages/junior/tests/integration/agent-invocation-concurrency.test.ts
index 2412071b60..b5cf833c31 100644
--- a/packages/junior/tests/integration/agent-invocation-concurrency.test.ts
+++ b/packages/junior/tests/integration/agent-invocation-concurrency.test.ts
@@ -8,34 +8,28 @@ import {
   MAX_ACTIVE_AGENT_INVOCATIONS_PER_PARENT,
 } from "@/chat/agent-invocations/store";
 import {
-  createAgentInvocationWorker,
   enqueueAgentInvocation,
-  routeAgentInvocationWork,
   createAndEnqueueAgentInvocation,
 } from "@/chat/agent-invocations/work";
 import type { CreateAgentInvocationInput } from "@/chat/agent-invocations/types";
-import { startAgentInvocationsForChildInput } from "@/chat/agent-invocations/result";
 import { EVENT_SYSTEM_ACTOR } from "@/chat/events/actor";
 import { getEventCatalog } from "@/chat/events/runtime-catalog";
 import { runTimerWatchHeartbeat } from "@/chat/events/timers";
 import { getAutomatedTurnLimitState } from "@/chat/services/automated-turn-limit";
 import { createEventTools } from "@/chat/tools/events";
-import { skipInputBeforeConversationStop } from "@/chat/conversations/session-stop";
 import { getConversationEventStore } from "@/chat/db";
 import { context, execute } from "../fixtures/event-automations";
 import { migrateSchema } from "@/chat/conversations/sql/migrations";
 import { createSqlStore } from "@/chat/conversations/sql/store";
-import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
+import { disconnectStateAdapter } from "@/chat/state/adapter";
 import { processConversationQueueMessage } from "@/chat/task-execution/vercel-callback";
 import { createListAgentsTool } from "@/chat/tools/runtime/list-agents";
 import { createStopAgentTool } from "@/chat/tools/runtime/stop-agent";
-import {
-  createConversationWorkQueueTestAdapter,
-  deferred,
-} from "../fixtures/conversation-work";
+import { deferred } from "../fixtures/conversation-work";
 import { createConfiguredJuniorSqlFixture } from "../fixtures/sql";
 import { createModelAgentRunnerForRun } from "../fixtures/agent-runner";
 import { createModelStream } from "../fixtures/model-stream";
+import { createConversationWorkHarness } from "../fixtures/conversation-work";
 
 const parentConversationId = "local:test:agent-invocation-concurrency";
 const destination = {
@@ -70,21 +64,27 @@ async function createHarness(streamForRun: (request: AgentRun) => StreamFn) {
     nowMs: 1_000,
     source: "local",
   });
-  const state = getStateAdapter();
-  await state.connect();
-  const queue = createConversationWorkQueueTestAdapter();
   const agentRunner = createModelAgentRunnerForRun(streamForRun);
   const run = vi.spyOn(agentRunner, "run");
-  const route = routeAgentInvocationWork({
-    // The same order as production conversation work.
-    fallbackWorker: async (workContext) =>
-      (await skipInputBeforeConversationStop(workContext)) ??
-      (await startAgentInvocationsForChildInput(workContext, {
-        queue,
-        state,
-      })) ?? { status: "completed" as const },
-    invocationWorker: createAgentInvocationWorker(agentRunner),
-  });
+  const {
+    queue,
+    run: route,
+    state,
+  } = await createConversationWorkHarness(agentRunner);
+  // This harness runs only child work. Each child result also wakes the
+  // parent, and agent-invocation-work covers that parent Turn.
+  const processNext = async () => {
+    const message = queue.takeMessage();
+    if (message.conversationId === parentConversationId) {
+      return;
+    }
+    await processConversationQueueMessage(message, {
+      conversationStore,
+      queue,
+      run: route,
+      state,
+    });
+  };
 
   return {
     async close() {
@@ -93,30 +93,12 @@ async function createHarness(streamForRun: (request: AgentRun) => StreamFn) {
     async drainQueuedWork() {
       while (queue.hasQueuedMessages()) {
         const batch = queue.queuedMessages();
-        await Promise.all(
-          batch.map(async () => {
-            await processConversationQueueMessage(queue.takeMessage(), {
-              conversationStore,
-              queue,
-              run: route,
-              state,
-            });
-          }),
-        );
+        await Promise.all(batch.map(processNext));
       }
     },
     processQueuedBatch: async () => {
       const batch = queue.queuedMessages();
-      await Promise.all(
-        batch.map(async () => {
-          await processConversationQueueMessage(queue.takeMessage(), {
-            conversationStore,
-            queue,
-            run: route,
-            state,
-          });
-        }),
-      );
+      await Promise.all(batch.map(processNext));
     },
     /** Send one parent message to an existing child, as messageAgent does. */
     message: async (args: {
@@ -499,7 +481,8 @@ describe("agent invocation identity and concurrency", () => {
           {
             conversation_id: parentConversationId,
             relation: "parent",
-            status: "idle",
+            // Both child results wait in the parent mailbox.
+            status: "pending",
             updated_at: expect.any(String),
           },
           { relation: "sibling", ...failingChild },
diff --git a/packages/junior/tests/integration/agent-invocation-message.test.ts b/packages/junior/tests/integration/agent-invocation-message.test.ts
index 2ac8ffe347..73cb8c7da5 100644
--- a/packages/junior/tests/integration/agent-invocation-message.test.ts
+++ b/packages/junior/tests/integration/agent-invocation-message.test.ts
@@ -1,9 +1,5 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { bindMessageAgent } from "@/chat/agent-invocations/message";
-import {
-  startAgentInvocationsForChildInput,
-  withAgentInvocationResultActors,
-} from "@/chat/agent-invocations/result";
 import {
   completeAgentInvocation,
   createAgentInvocation,
@@ -14,18 +10,17 @@ import {
 } from "@/chat/agent-invocations/store";
 import type { AgentInvocation } from "@/chat/agent-invocations/types";
 import type { AgentRun } from "@/chat/agent/types";
-import { resolveMailboxTurnWork } from "@/chat/task-execution/mailbox-turn";
 import { createSlackSource } from "@sentry/junior-plugin-api";
 import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
 import {
   getConversationWorkState,
   type InboundMessage,
 } from "@/chat/task-execution/store";
-import type { ConversationWorkerContext } from "@/chat/task-execution/worker";
 import { createListAgentsTool } from "@/chat/tools/runtime/list-agents";
 import { createMessageAgentTool } from "@/chat/tools/runtime/message-agent";
 import { stopChildAgentInvocations } from "@/chat/agent-invocations/stop";
-import { createConversationWorkQueueTestAdapter } from "../fixtures/conversation-work";
+import { neverRunAgentRunner } from "../fixtures/agent-runner";
+import { createConversationWorkHarness } from "../fixtures/conversation-work";
 import {
   prepareSlackParentConversation,
   slackDestination,
@@ -75,21 +70,20 @@ async function eventMessages(
   return work?.messages.filter((message) => message.source === "event") ?? [];
 }
 
-function context(
-  conversationId: string,
-  messages: InboundMessage[],
-): ConversationWorkerContext {
+/**
+ * Integration tests do not run the agent. This runner reads each run and
+ * stops it, so the work retries.
+ */
+function readRunsAgentRunner() {
+  const runs: AgentRun[] = [];
   return {
-    attempt: {
-      ack: async () => undefined,
-      conversationId,
-      drain: async () => [],
-      isFinalAttempt: false,
-      messages,
+    agentRunner: {
+      run: async (run: AgentRun): Promise<never> => {
+        runs.push(run);
+        throw new Error("stop after reading the run");
+      },
     },
-    checkIn: async () => true,
-    conversationId,
-    shouldYield: () => false,
+    runs,
   };
 }
 
@@ -101,9 +95,9 @@ describe("messageAgent", () => {
 
   it("gives a busy child more work and lets a child message its parent", async () => {
     const { fixture } = await prepareSlackParentConversation();
-    const queue = createConversationWorkQueueTestAdapter();
-    const state = getStateAdapter();
-    await state.connect();
+    const parent = readRunsAgentRunner();
+    const harness = await createConversationWorkHarness(parent.agentRunner);
+    const { queue, state } = harness;
     const mailbox = async (conversationId: string) =>
       (await getConversationWorkState({ conversationId, state }))?.messages ??
       [];
@@ -188,15 +182,13 @@ describe("messageAgent", () => {
       expect(parentMessages[0]!.input.text).toContain(
         "Found the bug in the retry path.",
       );
-      const turn = await withAgentInvocationResultActors(
-        (await resolveMailboxTurnWork(
-          context(slackParentConversationId, parentMessages),
-        ))!,
-      );
-      expect(turn).toMatchObject({
-        batch: [{ actor: parentActor }],
-        kind: "mailbox",
-      });
+      // The parent Turn runs after the Event burst window.
+      vi.spyOn(Date, "now").mockReturnValue(Date.now() + 60_000);
+      await harness.process(queue.takeMessageFor(slackParentConversationId));
+      vi.mocked(Date.now).mockRestore();
+      expect(parent.runs).toEqual([
+        expect.objectContaining({ actor: parentActor }),
+      ]);
 
       // The first work finishes. The queued message reports for both.
       await expect(
@@ -226,9 +218,9 @@ describe("messageAgent", () => {
 
   it("starts a new agent invocation when a grandchild messages a child, even after a stop", async () => {
     const { fixture } = await prepareSlackParentConversation();
-    const queue = createConversationWorkQueueTestAdapter();
-    const state = getStateAdapter();
-    await state.connect();
+    const runner = readRunsAgentRunner();
+    const harness = await createConversationWorkHarness(runner.agentRunner);
+    const { queue, state } = harness;
     try {
       const child = await createAgentInvocation({
         ...slackInvocationInput,
@@ -256,25 +248,20 @@ describe("messageAgent", () => {
         childConversationId: helper.childConversationId,
         parentConversationId: child.childConversationId,
       });
-      // A retried delivery starts the same agent invocation.
-      for (let attempt = 0; attempt < 2; attempt++) {
-        await expect(
-          startAgentInvocationsForChildInput(
-            context(child.childConversationId, messages),
-            { queue, state },
-          ),
-        ).resolves.toEqual({ status: "completed" });
-      }
-      await expect(
-        getAgentInvocation(
-          getAgentInvocationIdForMessage(messages[0]!.inboundMessageId),
-        ),
-      ).resolves.toMatchObject({
+      // The message starts and runs a new agent invocation on the child.
+      await harness.process(queue.takeMessageFor(child.childConversationId));
+      const started = await getAgentInvocation(
+        getAgentInvocationIdForMessage(messages[0]!.inboundMessageId),
+      );
+      expect(started).toMatchObject({
         childConversationId: child.childConversationId,
         input: expect.stringContaining("Halfway there."),
         parentConversationId: slackParentConversationId,
-        status: "pending",
+        status: "running",
       });
+      expect(runner.runs).toEqual([
+        expect.objectContaining({ runId: started!.invocationId }),
+      ]);
     } finally {
       await fixture.close();
     }
@@ -282,9 +269,8 @@ describe("messageAgent", () => {
 
   it("drops a grandchild message when the tree of its parent is stopped", async () => {
     const { fixture } = await prepareSlackParentConversation();
-    const queue = createConversationWorkQueueTestAdapter();
-    const state = getStateAdapter();
-    await state.connect();
+    const harness = await createConversationWorkHarness(neverRunAgentRunner());
+    const { queue, state } = harness;
     try {
       const child = await createAgentInvocation({
         ...slackInvocationInput,
@@ -327,11 +313,8 @@ describe("messageAgent", () => {
       const messages = await eventMessages(child.childConversationId);
       expect(messages).toHaveLength(1);
       await expect(
-        startAgentInvocationsForChildInput(
-          context(child.childConversationId, messages),
-          { queue, state },
-        ),
-      ).resolves.toEqual({ status: "completed" });
+        harness.process(queue.takeMessageFor(child.childConversationId)),
+      ).resolves.toMatchObject({ status: "completed" });
       await expect(
         getAgentInvocation(
           getAgentInvocationIdForMessage(messages[0]!.inboundMessageId),
diff --git a/packages/junior/tests/integration/agent-invocation-nested.test.ts b/packages/junior/tests/integration/agent-invocation-nested.test.ts
index 963fdabb2a..cc7808aa31 100644
--- a/packages/junior/tests/integration/agent-invocation-nested.test.ts
+++ b/packages/junior/tests/integration/agent-invocation-nested.test.ts
@@ -6,63 +6,18 @@ import {
   getAgentInvocationIdForResult,
   getAgentInvocationResultMessageId,
 } from "@/chat/agent-invocations/store";
-import {
-  enqueueAgentInvocationResult,
-  startAgentInvocationsForChildInput,
-} from "@/chat/agent-invocations/result";
+import { enqueueAgentInvocationResult } from "@/chat/agent-invocations/result";
 import type { TerminalAgentInvocation } from "@/chat/agent-invocations/types";
-import { createAgentInvocationWorker } from "@/chat/agent-invocations/work";
 import type { AgentRun } from "@/chat/agent/types";
-import { migrateSchema } from "@/chat/conversations/sql/migrations";
-import { createSqlStore } from "@/chat/conversations/sql/store";
-import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
-import {
-  getConversationWorkState,
-  type InboundMessage,
-} from "@/chat/task-execution/store";
-import type { ConversationWorkerContext } from "@/chat/task-execution/worker";
+import { disconnectStateAdapter } from "@/chat/state/adapter";
+import { getConversationWorkState } from "@/chat/task-execution/store";
 import { createListAgentsTool } from "@/chat/tools/runtime/list-agents";
-import { createSlackSource } from "@sentry/junior-plugin-api";
-import { createConversationWorkQueueTestAdapter } from "../fixtures/conversation-work";
-import { createConfiguredJuniorSqlFixture } from "../fixtures/sql";
-
-const slackParentConversationId = "slack:C123:1712345.0001";
-const slackDestination = {
-  platform: "slack",
-  teamId: "T123",
-  channelId: "C123",
-} as const;
-const slackInvocationInput = {
-  actor: {
-    platform: "slack",
-    teamId: "T123",
-    userId: "U123",
-  } as const,
-  destination: slackDestination,
-  input: "Summarize the durable task.",
-  parentConversationId: slackParentConversationId,
-  reasoningLevel: "medium" as const,
-};
-
-async function prepareSlackParentConversation() {
-  const fixture = createConfiguredJuniorSqlFixture();
-  await migrateSchema(fixture.sql);
-  const conversationStore = createSqlStore(fixture.sql);
-  await conversationStore.recordActivity({
-    conversationId: slackParentConversationId,
-    destination: slackDestination,
-    nowMs: 1_000,
-    sessionSource: createSlackSource({
-      channelId: "C123",
-      teamId: "T123",
-      threadTs: "1712345.0001",
-      visibility: "public",
-    }),
-    source: "slack",
-    visibility: "public",
-  });
-  return { fixture };
-}
+import { createConversationWorkHarness } from "../fixtures/conversation-work";
+import {
+  prepareSlackParentConversation,
+  slackInvocationInput,
+  slackParentConversationId,
+} from "../fixtures/agent-invocations";
 
 async function listAgents(conversationId: string) {
   return (await createListAgentsTool({ conversationId }).execute!({}, {}))
@@ -77,24 +32,16 @@ describe("nested agent invocations", () => {
 
   it("sends each nested child's report to its parent once its own children finish", async () => {
     const { fixture } = await prepareSlackParentConversation();
-    const queue = createConversationWorkQueueTestAdapter();
-    const state = getStateAdapter();
-    await state.connect();
-    const context = (
-      conversationId: string,
-      messages: InboundMessage[] = [],
-    ): ConversationWorkerContext => ({
-      attempt: {
-        ack: async () => undefined,
-        conversationId,
-        drain: async () => [],
-        isFinalAttempt: false,
-        messages,
+    // Integration tests do not run the agent. The runner reads each run and
+    // stops it, so its work retries until the test finishes the invocation.
+    const runs: AgentRun[] = [];
+    const harness = await createConversationWorkHarness({
+      run: async (run) => {
+        runs.push(run);
+        throw new Error("stop after reading the run");
       },
-      checkIn: async () => true,
-      conversationId,
-      shouldYield: () => false,
     });
+    const { queue, state } = harness;
     const mailbox = async (conversationId: string) =>
       (await getConversationWorkState({ conversationId, state }))?.messages ??
       [];
@@ -104,29 +51,28 @@ describe("nested agent invocations", () => {
         result,
         status: "completed",
       })) as TerminalAgentInvocation;
-    // Send one result to its parent child, as the worker does, and return the
-    // agent invocation that the result starts there.
-    const sendToChild = async (result: TerminalAgentInvocation) => {
-      await enqueueAgentInvocationResult(result, { queue, state });
-      const messages = (await mailbox(result.parentConversationId)).filter(
-        (message) =>
-          message.inboundMessageId ===
-          getAgentInvocationResultMessageId(result.invocationId),
-      );
-      expect(messages).toHaveLength(1);
-      // A retried delivery starts the same agent invocation.
-      for (let attempt = 0; attempt < 2; attempt++) {
-        await expect(
-          startAgentInvocationsForChildInput(
-            context(result.parentConversationId, messages),
-            { queue, state },
-          ),
-        ).resolves.toEqual({ status: "completed" });
-      }
+    // Production work gives a result to the parent child. That starts a new
+    // agent invocation there and runs it.
+    const deliver = async (result: TerminalAgentInvocation) => {
+      await expect(
+        harness.process(queue.takeMessageFor(result.parentConversationId)),
+      ).resolves.toMatchObject({ status: "failed" });
       return (await getAgentInvocation(
         getAgentInvocationIdForResult(result.invocationId),
       ))!;
     };
+    // Finish the run of one agent invocation. Its retry wake reports the
+    // result to the parent.
+    const finish = async (
+      invocation: { childConversationId: string; invocationId: string },
+      result: string,
+    ) => {
+      const terminal = await complete(invocation.invocationId, result);
+      await expect(
+        harness.process(queue.takeMessageFor(invocation.childConversationId)),
+      ).resolves.toMatchObject({ status: "completed" });
+      return terminal;
+    };
     try {
       const child = await createAgentInvocation({
         ...slackInvocationInput,
@@ -143,57 +89,21 @@ describe("nested agent invocations", () => {
         parentConversationId: helper.childConversationId,
       });
 
-      // With maxAgentDepth 1, depth 1 can spawn and depth 2 cannot. Both read
-      // the root Location.
-      const runs: AgentRun[] = [];
-      const worker = createAgentInvocationWorker(
-        {
-          run: async (run) => {
-            runs.push(run);
-            throw new Error("stop after reading the run");
-          },
-        },
-        { maxAgentDepth: 1 },
-      );
-      for (const invocation of [child, helper]) {
-        await expect(
-          worker(
-            context(invocation.childConversationId),
-            invocation.invocationId,
-          ),
-        ).rejects.toThrow("stop after reading the run");
-      }
-      expect(runs.map((run) => run.disabledFeatures)).toEqual([
-        ["handoff", "interactive-auth"],
-        ["handoff", "interactive-auth", "subagents"],
-      ]);
-      expect(runs[1]?.location).toMatchObject({
-        provider: "slack",
-        channelId: "C123",
-      });
-
       // A child that finishes while its own child runs does not report yet.
       for (const invocation of [child, helper]) {
         await expect(
           complete(invocation.invocationId, "Started a helper."),
         ).resolves.toMatchObject({ parentMailboxStatus: "replaced" });
       }
-
       const grandResult = await complete(
         grandHelper.invocationId,
         "Grand helper fact: 42.",
       );
       expect(grandResult).toMatchObject({ parentMailboxStatus: "pending" });
-      const helperContinuation = await sendToChild(grandResult);
-      expect(helperContinuation).toMatchObject({
-        actor: slackInvocationInput.actor,
-        childConversationId: helper.childConversationId,
-        input: expect.stringContaining("Grand helper fact: 42."),
-        parentConversationId: child.childConversationId,
-        status: "pending",
-      });
+      await enqueueAgentInvocationResult(grandResult, { queue, state });
+
       // listAgents shows the state of open work. A `replaced` child still
-      // has work, and an agent invocation that receives a result is not a task.
+      // has work.
       const listedChild = (childConversationId: string) => ({
         conversation_id: childConversationId,
         relation: "child",
@@ -208,20 +118,44 @@ describe("nested agent invocations", () => {
         listedChild(helper.childConversationId),
       ]);
 
+      // The result runs on the helper as the Actor that started the tree,
+      // with the root Location.
+      const helperContinuation = await deliver(grandResult);
+      expect(helperContinuation).toMatchObject({
+        actor: slackInvocationInput.actor,
+        childConversationId: helper.childConversationId,
+        input: expect.stringContaining("Grand helper fact: 42."),
+        parentConversationId: child.childConversationId,
+        status: "running",
+      });
+      expect(runs).toEqual([
+        expect.objectContaining({
+          actor: slackInvocationInput.actor,
+          conversationId: helper.childConversationId,
+          disabledFeatures: ["handoff", "interactive-auth", "subagents"],
+          location: expect.objectContaining({
+            provider: "slack",
+            channelId: "C123",
+          }),
+        }),
+      ]);
+      // An agent invocation that receives a result is not a new task.
+      await expect(listAgents(child.childConversationId)).resolves.toEqual([
+        { ...listedChild(helper.childConversationId), status: "running" },
+      ]);
+
       // Each child reports once, from the agent invocation that received the
       // last result of its own children.
-      const helperResult = await complete(
-        helperContinuation.invocationId,
-        "Helper fact: 42.",
-      );
-      expect(helperResult).toMatchObject({ parentMailboxStatus: "pending" });
-      const childContinuation = await sendToChild(helperResult);
-      const childResult = await complete(
-        childContinuation.invocationId,
-        "Final answer: 42.",
-      );
-      expect(childResult).toMatchObject({ parentMailboxStatus: "pending" });
-      await enqueueAgentInvocationResult(childResult, { queue, state });
+      const helperResult = await finish(helperContinuation, "Helper fact: 42.");
+      const childContinuation = await deliver(helperResult);
+      expect(childContinuation).toMatchObject({
+        input: expect.stringContaining("Helper fact: 42."),
+        parentConversationId: slackParentConversationId,
+      });
+      await finish(childContinuation, "Final answer: 42.");
+      await expect(
+        getAgentInvocation(childContinuation.invocationId),
+      ).resolves.toMatchObject({ parentMailboxStatus: "appended" });
       expect(await mailbox(slackParentConversationId)).toEqual([
         expect.objectContaining({
           inboundMessageId: getAgentInvocationResultMessageId(
diff --git a/packages/junior/tests/integration/agent-invocation-spawn.test.ts b/packages/junior/tests/integration/agent-invocation-spawn.test.ts
index bd64c4d3df..01486b3bff 100644
--- a/packages/junior/tests/integration/agent-invocation-spawn.test.ts
+++ b/packages/junior/tests/integration/agent-invocation-spawn.test.ts
@@ -4,20 +4,13 @@ import {
   createAgentInvocation,
   getAgentInvocation,
   getAgentInvocationId,
-  getAgentInvocationIdForResult,
   getAgentInvocationMessageId,
   getAgentInvocationResultMessageId,
   getAgentInvocationTurnId,
 } from "@/chat/agent-invocations/store";
-import {
-  enqueueAgentInvocationResult,
-  startAgentInvocationsForChildInput,
-} from "@/chat/agent-invocations/result";
+import { enqueueAgentInvocationResult } from "@/chat/agent-invocations/result";
 import type { TerminalAgentInvocation } from "@/chat/agent-invocations/types";
-import type { InboundMessage } from "@/chat/task-execution/store";
-import type { ConversationWorkerContext } from "@/chat/task-execution/worker";
 import { getConversationWorkState } from "@/chat/task-execution/store";
-import { createAgentInvocationWorker } from "@/chat/agent-invocations/work";
 import { bindSpawnAgent } from "@/chat/agent-invocations/spawn";
 import { createSpawnAgentTool } from "@/chat/tools/runtime/spawn-agent";
 import type { AgentRun } from "@/chat/agent/types";
@@ -248,153 +241,6 @@ describe("agent invocation spawn and result delivery", () => {
     }
   });
 
-  it("sends each nested child's report to its parent once its own children finish", async () => {
-    const { fixture } = await prepareSlackParentConversation();
-    const queue = createConversationWorkQueueTestAdapter();
-    const state = getStateAdapter();
-    await state.connect();
-    const context = (
-      conversationId: string,
-      messages: InboundMessage[] = [],
-    ): ConversationWorkerContext => ({
-      attempt: {
-        ack: async () => undefined,
-        conversationId,
-        drain: async () => [],
-        isFinalAttempt: false,
-        messages,
-      },
-      checkIn: async () => true,
-      conversationId,
-      shouldYield: () => false,
-    });
-    const mailbox = async (conversationId: string) =>
-      (await getConversationWorkState({ conversationId, state }))?.messages ??
-      [];
-    const complete = async (invocationId: string, result: string) =>
-      (await completeAgentInvocation({
-        invocationId,
-        result,
-        status: "completed",
-      })) as TerminalAgentInvocation;
-    // Send one result to its parent child, as the worker does, and return the
-    // agent invocation that the result starts there.
-    const sendToChild = async (result: TerminalAgentInvocation) => {
-      await enqueueAgentInvocationResult(result, { queue, state });
-      const messages = (await mailbox(result.parentConversationId)).filter(
-        (message) =>
-          message.inboundMessageId ===
-          getAgentInvocationResultMessageId(result.invocationId),
-      );
-      expect(messages).toHaveLength(1);
-      // A retried delivery starts the same agent invocation.
-      for (let attempt = 0; attempt < 2; attempt++) {
-        await expect(
-          startAgentInvocationsForChildInput(
-            context(result.parentConversationId, messages),
-            { queue, state },
-          ),
-        ).resolves.toEqual({ status: "completed" });
-      }
-      return (await getAgentInvocation(
-        getAgentInvocationIdForResult(result.invocationId),
-      ))!;
-    };
-    try {
-      const child = await createAgentInvocation({
-        ...slackInvocationInput,
-        idempotencyKey: "nested-child",
-      });
-      const helper = await createAgentInvocation({
-        ...slackInvocationInput,
-        idempotencyKey: "nested-helper",
-        parentConversationId: child.childConversationId,
-      });
-      const grandHelper = await createAgentInvocation({
-        ...slackInvocationInput,
-        idempotencyKey: "nested-grand-helper",
-        parentConversationId: helper.childConversationId,
-      });
-
-      // With maxAgentDepth 1, depth 1 can spawn and depth 2 cannot. Both read
-      // the root Location.
-      const runs: AgentRun[] = [];
-      const worker = createAgentInvocationWorker(
-        {
-          run: async (run) => {
-            runs.push(run);
-            throw new Error("stop after reading the run");
-          },
-        },
-        { maxAgentDepth: 1 },
-      );
-      for (const invocation of [child, helper]) {
-        await expect(
-          worker(
-            context(invocation.childConversationId),
-            invocation.invocationId,
-          ),
-        ).rejects.toThrow("stop after reading the run");
-      }
-      expect(runs.map((run) => run.disabledFeatures)).toEqual([
-        ["handoff", "interactive-auth"],
-        ["handoff", "interactive-auth", "subagents"],
-      ]);
-      expect(runs[1]?.location).toMatchObject({
-        provider: "slack",
-        channelId: "C123",
-      });
-
-      // A child that finishes while its own child runs does not report yet.
-      for (const invocation of [child, helper]) {
-        await expect(
-          complete(invocation.invocationId, "Started a helper."),
-        ).resolves.toMatchObject({ parentMailboxStatus: "replaced" });
-      }
-
-      const grandResult = await complete(
-        grandHelper.invocationId,
-        "Grand helper fact: 42.",
-      );
-      expect(grandResult).toMatchObject({ parentMailboxStatus: "pending" });
-      const helperContinuation = await sendToChild(grandResult);
-      expect(helperContinuation).toMatchObject({
-        actor: slackInvocationInput.actor,
-        childConversationId: helper.childConversationId,
-        input: expect.stringContaining("Grand helper fact: 42."),
-        parentConversationId: child.childConversationId,
-        status: "pending",
-      });
-
-      // Each child reports once, from the agent invocation that received the
-      // last result of its own children.
-      const helperResult = await complete(
-        helperContinuation.invocationId,
-        "Helper fact: 42.",
-      );
-      expect(helperResult).toMatchObject({ parentMailboxStatus: "pending" });
-      const childContinuation = await sendToChild(helperResult);
-      const childResult = await complete(
-        childContinuation.invocationId,
-        "Final answer: 42.",
-      );
-      expect(childResult).toMatchObject({ parentMailboxStatus: "pending" });
-      await enqueueAgentInvocationResult(childResult, { queue, state });
-      expect(await mailbox(slackParentConversationId)).toEqual([
-        expect.objectContaining({
-          inboundMessageId: getAgentInvocationResultMessageId(
-            childContinuation.invocationId,
-          ),
-          input: expect.objectContaining({
-            text: expect.stringContaining("Final answer: 42."),
-          }),
-        }),
-      ]);
-    } finally {
-      await fixture.close();
-    }
-  });
-
   it("repairs the durable creation-to-mailbox crash gap once", async () => {
     const { fixture } = await prepareParentConversation();
     const queue = createConversationWorkQueueTestAdapter();
diff --git a/packages/junior/tests/integration/agent-invocation-stop.test.ts b/packages/junior/tests/integration/agent-invocation-stop.test.ts
index ccafbe9b42..34fd121a5b 100644
--- a/packages/junior/tests/integration/agent-invocation-stop.test.ts
+++ b/packages/junior/tests/integration/agent-invocation-stop.test.ts
@@ -1,9 +1,6 @@
 import { afterEach, describe, expect, it, vi } from "vitest";
 import type { AgentRun } from "@/chat/agent/types";
-import {
-  enqueueAgentInvocationResult,
-  startAgentInvocationsForChildInput,
-} from "@/chat/agent-invocations/result";
+import { enqueueAgentInvocationResult } from "@/chat/agent-invocations/result";
 import {
   completeAgentInvocation,
   createAgentInvocation,
@@ -17,18 +14,15 @@ import type {
   CreateAgentInvocationInput,
   TerminalAgentInvocation,
 } from "@/chat/agent-invocations/types";
-import {
-  createAgentInvocationWorker,
-  createAndEnqueueAgentInvocation,
-  routeAgentInvocationWork,
-} from "@/chat/agent-invocations/work";
+import { createAndEnqueueAgentInvocation } from "@/chat/agent-invocations/work";
+import { stopConversationTurn } from "@/chat/conversations/stop";
 import type { AgentRunner } from "@/chat/runtime/agent-runner";
-import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
+import { neverRunAgentRunner } from "../fixtures/agent-runner";
+import { disconnectStateAdapter } from "@/chat/state/adapter";
 import {
   getConversationWorkState,
   hasConversationStop,
 } from "@/chat/task-execution/store";
-import type { ConversationWorkerContext } from "@/chat/task-execution/worker";
 import { createListAgentsTool } from "@/chat/tools/runtime/list-agents";
 import { createStopAgentTool } from "@/chat/tools/runtime/stop-agent";
 import {
@@ -39,7 +33,7 @@ import {
   prepareParentConversation,
 } from "../fixtures/agent-invocations";
 import {
-  createConversationWorkQueueTestAdapter,
+  createConversationWorkHarness,
   deferred,
 } from "../fixtures/conversation-work";
 
@@ -74,36 +68,24 @@ async function waitForAbort(run: AgentRun): Promise<never> {
   throw signal.reason;
 }
 
-async function createHarness() {
+/**
+ * Compose production conversation work for a local parent. Each test runs
+ * child work from the queue, as the queue callback does.
+ */
+async function createHarness(agentRunner: AgentRunner) {
   const { conversationStore, fixture } = await prepareParentConversation();
-  const state = getStateAdapter();
-  await state.connect();
-  const queue = createConversationWorkQueueTestAdapter();
-  const mailbox = async (conversationId: string) =>
-    (await getConversationWorkState({ conversationId, state }))?.messages ?? [];
+  const work = await createConversationWorkHarness(agentRunner);
+  const { queue, state } = work;
   return {
     close: async () => await fixture.close(),
-    /** Build the worker context for the child's queued input. */
-    async context(
-      conversationId: string,
-      signal?: AbortSignal,
-    ): Promise<ConversationWorkerContext> {
-      const messages = await mailbox(conversationId);
-      return {
-        attempt: {
-          ack: async () => undefined,
-          conversationId,
-          drain: async () => [],
-          isFinalAttempt: false,
-          messages,
-        },
-        checkIn: async () => true,
-        conversationId,
-        shouldYield: () => false,
-        ...(signal ? { stopSignal: () => signal } : undefined),
-      };
-    },
-    mailbox,
+    conversationStore,
+    mailbox: async (conversationId: string) =>
+      (await getConversationWorkState({ conversationId, state }))?.messages ??
+      [],
+    process: work.process,
+    /** Run the next queued wake of one Conversation. */
+    processNext: async (conversationId: string) =>
+      await work.process(queue.takeMessageFor(conversationId)),
     queue,
     spawn: async (
       overrides: Partial<CreateAgentInvocationInput> & {
@@ -115,11 +97,27 @@ async function createHarness() {
         { conversationStore, queue, state },
       ),
     state,
-    worker: (agentRunner: AgentRunner) =>
-      createAgentInvocationWorker(agentRunner, {
-        enqueueResult: async (invocation) =>
-          await enqueueAgentInvocationResult(invocation, { queue, state }),
-      }),
+  };
+}
+
+/** Run each child until a stop aborts it, and report when it starts. */
+function abortableAgentRunner() {
+  let started = deferred<void>();
+  const runs: AgentRun[] = [];
+  return {
+    agentRunner: {
+      run: async (run: AgentRun) => {
+        runs.push(run);
+        started.resolve(undefined);
+        return await waitForAbort(run);
+      },
+    } satisfies AgentRunner,
+    runs,
+    /** Wait for the next run to start. */
+    started: async () => {
+      await started.promise;
+      started = deferred<void>();
+    },
   };
 }
 
@@ -130,26 +128,16 @@ describe("stopping child agents", () => {
   });
 
   it("stops a running child without a result and lets a message start new work", async () => {
-    const harness = await createHarness();
+    const runner = abortableAgentRunner();
+    const harness = await createHarness(runner.agentRunner);
     try {
       const running = await harness.spawn({
         agentName: "researcher",
         idempotencyKey: "stop-running",
         input: "long task",
       });
-      const started = deferred<void>();
-      // Stands in for the worker's poll of the Conversation stop request.
-      const stopRequest = new AbortController();
-      const processing = harness.worker({
-        run: async (run) => {
-          started.resolve(undefined);
-          return await waitForAbort(run);
-        },
-      })(
-        await harness.context(running.childConversationId, stopRequest.signal),
-        running.invocationId,
-      );
-      await started.promise;
+      const processing = harness.processNext(running.childConversationId);
+      await runner.started();
 
       await expect(stopAgent(running.childConversationId)).resolves.toEqual({
         conversation_id: running.childConversationId,
@@ -166,8 +154,8 @@ describe("stopping child agents", () => {
           state: harness.state,
         }),
       ).resolves.toBe(true);
-      stopRequest.abort(new Error("Conversation work stopped"));
-      await expect(processing).resolves.toEqual({ status: "completed" });
+      // The worker sees the stop request and aborts the run.
+      await expect(processing).resolves.toMatchObject({ status: "completed" });
 
       await expect(
         getAgentInvocation(running.invocationId),
@@ -237,12 +225,8 @@ describe("stopping child agents", () => {
   });
 
   it("never runs a stopped pending child and reports other stops to the parent", async () => {
-    const harness = await createHarness();
-    const neverRun: AgentRunner = {
-      run: async () => {
-        throw new Error("A stopped child must not run");
-      },
-    };
+    const runner = abortableAgentRunner();
+    const harness = await createHarness(runner.agentRunner);
     try {
       const pending = await harness.spawn({
         idempotencyKey: "stop-pending",
@@ -255,33 +239,20 @@ describe("stopping child agents", () => {
         conversation_id: pending.childConversationId,
         stopped: true,
       });
-      await expect(
-        harness.worker(neverRun)(
-          await harness.context(pending.childConversationId),
-          pending.invocationId,
-        ),
-      ).resolves.toEqual({ status: "completed" });
+      // The stopped child takes its wake without a run. A redelivery finds
+      // no work.
+      const wake = harness.queue.takeMessageFor(pending.childConversationId);
+      await expect(harness.process(wake)).resolves.toMatchObject({
+        status: "completed",
+      });
+      await expect(harness.process(wake)).resolves.toMatchObject({
+        status: "no_work",
+      });
+      expect(runner.runs).toEqual([]);
       await expect(
         getAgentInvocation(pending.invocationId),
       ).resolves.toMatchObject({ status: "stopped" });
 
-      // A stop request can outlive its run. The stopped agent invocation
-      // handles the next empty wake.
-      const invocationWorker = vi.fn(async () => ({
-        status: "completed" as const,
-      }));
-      const wake = await harness.context(pending.childConversationId);
-      await routeAgentInvocationWork({
-        fallbackWorker: async () => {
-          throw new Error("Child work reached the fallback worker");
-        },
-        invocationWorker,
-      })({ ...wake, attempt: { ...wake.attempt, messages: [] } });
-      expect(invocationWorker).toHaveBeenCalledWith(
-        expect.anything(),
-        pending.invocationId,
-      );
-
       // A stop that does not come from stopAgent still reaches the parent,
       // and it also stops the child agents of the child.
       const external = await harness.spawn({
@@ -293,17 +264,17 @@ describe("stopping child agents", () => {
         idempotencyKey: "stop-external-helper",
         parentConversationId: external.childConversationId,
       });
-      const stopRequest = new AbortController();
-      stopRequest.abort(new Error("Conversation work stopped"));
+      const processing = harness.processNext(external.childConversationId);
+      await runner.started();
       await expect(
-        harness.worker({ run: waitForAbort })(
-          await harness.context(
-            external.childConversationId,
-            stopRequest.signal,
-          ),
-          external.invocationId,
-        ),
-      ).resolves.toEqual({ status: "completed" });
+        stopConversationTurn({
+          conversationId: external.childConversationId,
+          conversationStore: harness.conversationStore,
+          queue: harness.queue,
+          state: harness.state,
+        }),
+      ).resolves.toMatchObject({ status: "requested" });
+      await expect(processing).resolves.toMatchObject({ status: "completed" });
       await expect(
         getAgentInvocation(external.invocationId),
       ).resolves.toMatchObject({
@@ -341,7 +312,7 @@ describe("stopping child agents", () => {
   });
 
   it("stops every child agent below the child and drops their results", async () => {
-    const harness = await createHarness();
+    const harness = await createHarness(neverRunAgentRunner());
     try {
       const child = await createAgentInvocation({
         ...invocationInput,
@@ -402,14 +373,12 @@ describe("stopping child agents", () => {
         status: "completed",
       });
       // The child already had the result in its mailbox. It starts no work.
-      const childContext = await harness.context(child.childConversationId);
-      expect(childContext.attempt.messages).toHaveLength(1);
       await expect(
-        startAgentInvocationsForChildInput(childContext, {
-          queue: harness.queue,
-          state: harness.state,
-        }),
-      ).resolves.toEqual({ status: "completed" });
+        harness.mailbox(child.childConversationId),
+      ).resolves.toHaveLength(1);
+      await expect(
+        harness.processNext(child.childConversationId),
+      ).resolves.toMatchObject({ status: "completed" });
       await expect(
         getAgentInvocation(
           getAgentInvocationIdForResult(doneHelper.invocationId),
diff --git a/packages/junior/tests/integration/agent-invocation-work.test.ts b/packages/junior/tests/integration/agent-invocation-work.test.ts
index 1bf2a2135f..07d5638059 100644
--- a/packages/junior/tests/integration/agent-invocation-work.test.ts
+++ b/packages/junior/tests/integration/agent-invocation-work.test.ts
@@ -4,31 +4,33 @@ import { createApp, defineJuniorPlugins } from "@/app";
 import type { PiMessage } from "@/chat/pi/messages";
 import {
   completeAgentInvocation,
+  getActiveAgentInvocationForConversation,
   getAgentInvocation,
   getAgentInvocationMessageId,
   getAgentInvocationResultMessageId,
   getAgentInvocationTurnId,
 } from "@/chat/agent-invocations/store";
-import {
-  enqueueAgentInvocationResult,
-  withAgentInvocationResultActors,
-} from "@/chat/agent-invocations/result";
+import { withAgentInvocationResultActors } from "@/chat/agent-invocations/result";
 import { EVENT_MESSAGE_AUTHOR, EVENT_SYSTEM_ACTOR } from "@/chat/events/actor";
-import { getConversationWorkState } from "@/chat/task-execution/store";
+import { getEventCatalog } from "@/chat/events/runtime-catalog";
+import { runTimerWatchHeartbeat } from "@/chat/events/timers";
+import { createEventTools } from "@/chat/tools/events";
+import { createStopAgentTool } from "@/chat/tools/runtime/stop-agent";
 import {
-  createAgentInvocationWorker,
-  routeAgentInvocationWork,
-  createAndEnqueueAgentInvocation,
-} from "@/chat/agent-invocations/work";
+  CONVERSATION_WORK_MAX_RETRIES,
+  getConversationWorkState,
+} from "@/chat/task-execution/store";
+import { createAndEnqueueAgentInvocation } from "@/chat/agent-invocations/work";
 import { loadProjection } from "@/chat/conversations/projection";
 import { getConversationEventStore } from "@/chat/db";
-import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter";
+import { disconnectStateAdapter } from "@/chat/state/adapter";
 import { processConversationQueueMessage } from "@/chat/task-execution/vercel-callback";
 import {
   getTurnRecord,
   upsertTurnRecord,
 } from "@/chat/task-execution/turn-cursor";
-import { createConversationWorkQueueTestAdapter } from "../fixtures/conversation-work";
+import { createConversationWorkHarness } from "../fixtures/conversation-work";
+import { context, execute } from "../fixtures/event-automations";
 import {
   createModelAgentRunner,
   neverRunAgentRunner,
@@ -64,9 +66,24 @@ describe("agent invocation conversation work", () => {
   it("reads the parent Location and visibility without delivering child output", async () => {
     const { conversationStore, fixture } =
       await prepareSlackParentConversation();
-    const queue = createConversationWorkQueueTestAdapter();
-    const state = getStateAdapter();
-    await state.connect();
+    let observedVisibility: ReturnType<typeof getCurrentConversationPrivacy>;
+    const agentRunner = createModelAgentRunner(
+      createModelStream([
+        {
+          type: "text",
+          text: "Durable child result",
+          onRequest: () => {
+            observedVisibility = getCurrentConversationPrivacy();
+          },
+        },
+      ]),
+    );
+    const run = vi.spyOn(agentRunner, "run");
+    const {
+      queue,
+      run: route,
+      state,
+    } = await createConversationWorkHarness(agentRunner);
     try {
       const created = await createAndEnqueueAgentInvocation(
         {
@@ -90,29 +107,6 @@ describe("agent invocation conversation work", () => {
       expect(child).not.toHaveProperty("location");
       expect(child).not.toHaveProperty("sessionSource");
       expect(child).not.toHaveProperty("visibility");
-      let observedVisibility: ReturnType<typeof getCurrentConversationPrivacy>;
-      const agentRunner = createModelAgentRunner(
-        createModelStream([
-          {
-            type: "text",
-            text: "Durable child result",
-            onRequest: () => {
-              observedVisibility = getCurrentConversationPrivacy();
-            },
-          },
-        ]),
-      );
-      const run = vi.spyOn(agentRunner, "run");
-      const fallbackWorker = vi.fn(async () => ({
-        status: "completed" as const,
-      }));
-      const route = routeAgentInvocationWork({
-        fallbackWorker,
-        invocationWorker: createAgentInvocationWorker(agentRunner, {
-          enqueueResult: async (invocation) =>
-            await enqueueAgentInvocationResult(invocation, { queue, state }),
-        }),
-      });
       const queueMessage = queue.takeMessage();
 
       await expect(
@@ -310,7 +304,6 @@ describe("agent invocation conversation work", () => {
       });
       expect(run.mock.calls[0]?.[0].delivery).toBeUndefined();
       expect(observedVisibility).toBe("public");
-      expect(fallbackWorker).not.toHaveBeenCalled();
     } finally {
       await fixture.close();
     }
@@ -346,9 +339,13 @@ describe("agent invocation conversation work", () => {
       terminal,
     }) => {
       const { conversationStore, fixture } = await prepareParentConversation();
-      const queue = createConversationWorkQueueTestAdapter();
-      const state = getStateAdapter();
-      await state.connect();
+      const {
+        queue,
+        run: route,
+        state,
+      } = await createConversationWorkHarness(
+        createModelAgentRunner(createModelStream([output])),
+      );
       try {
         const created = await createAndEnqueueAgentInvocation(
           {
@@ -357,12 +354,6 @@ describe("agent invocation conversation work", () => {
           },
           { conversationStore, queue, state },
         );
-        const route = routeAgentInvocationWork({
-          fallbackWorker: vi.fn(async () => ({ status: "completed" as const })),
-          invocationWorker: createAgentInvocationWorker(
-            createModelAgentRunner(createModelStream([output])),
-          ),
-        });
 
         await expect(
           processConversationQueueMessage(queue.takeMessage(), {
@@ -408,58 +399,102 @@ describe("agent invocation conversation work", () => {
     },
   );
 
-  it("runs an approve plugin tool with no action review and resumes the yielded child", async () => {
-    const markDone = vi.fn((_input: { itemId: string }) => ({}));
-    await createApp({
-      plugins: defineJuniorPlugins([
-        defineJuniorPlugin({
-          manifest: {
-            name: "upkeep",
-            displayName: "Upkeep",
-            description: "Upkeep test tools",
-          },
-          hooks: {
-            tools: () => ({
-              markDone: definePluginTool({
-                // The tool checks in code that its effect is safe, so action
-                // review does not run, also in a child agent.
-                approvalMode: "approve",
-                exposure: "direct",
-                annotations: {
-                  destructiveHint: true,
-                  idempotentHint: true,
-                  openWorldHint: true,
-                  readOnlyHint: false,
-                },
-                description: "Mark one item done.",
-                inputSchema: z.object({ itemId: z.string() }),
-                outputSchema: pluginToolOutputSchema,
-                execute: markDone,
+  // A child watch event runs with an event Source. When its Turn yields, the
+  // next wake must still reach the agent invocation. Production once sent it
+  // to the root Event Turn worker, which needs a Destination that a child
+  // does not have. That also holds when the agent invocation ended or was
+  // stopped while its Turn stayed paused. A resume wake has no messages, but
+  // its last try is still final, so a resume that keeps failing tells the
+  // parent.
+  const resumed = { result: "Resumed child result", status: "completed" };
+  it.each([
+    {
+      name: "a parent task",
+      watchEvent: false,
+      endBeforeWake: false,
+      runs: 2,
+      expected: resumed,
+    },
+    {
+      name: "a watch event",
+      watchEvent: true,
+      endBeforeWake: false,
+      runs: 3,
+      expected: resumed,
+    },
+    {
+      name: "an ended watch event",
+      watchEvent: true,
+      endBeforeWake: true,
+      runs: 2,
+      expected: { parentMailboxStatus: "appended", status: "failed" },
+    },
+    {
+      name: "a stopped child",
+      watchEvent: false,
+      stopBeforeWake: true,
+      runs: 1,
+      expected: { parentMailboxStatus: "skipped", status: "stopped" },
+    },
+    {
+      name: "a resume that keeps failing",
+      watchEvent: false,
+      // The last try is final, so it fails the invocation, not the work.
+      failedResumes: CONVERSATION_WORK_MAX_RETRIES - 1,
+      runs: CONVERSATION_WORK_MAX_RETRIES + 1,
+      expected: {
+        errorMessage: "resume unavailable",
+        parentMailboxStatus: "appended",
+        status: "failed",
+      },
+    },
+  ])(
+    "gives the wake after a yielded child invocation for $name to that invocation",
+    async ({
+      endBeforeWake,
+      expected,
+      failedResumes = 0,
+      runs,
+      stopBeforeWake = false,
+      watchEvent,
+    }) => {
+      const clock = vi.spyOn(Date, "now").mockReturnValue(1_800_000_000_000);
+      const markDone = vi.fn((_input: { itemId: string }) => ({}));
+      await createApp({
+        plugins: defineJuniorPlugins([
+          defineJuniorPlugin({
+            manifest: {
+              name: "upkeep",
+              displayName: "Upkeep",
+              description: "Upkeep test tools",
+            },
+            hooks: {
+              tools: () => ({
+                markDone: definePluginTool({
+                  // The tool checks in code that its effect is safe, so action
+                  // review does not run, also in a child agent.
+                  approvalMode: "approve",
+                  exposure: "direct",
+                  annotations: {
+                    destructiveHint: true,
+                    idempotentHint: true,
+                    openWorldHint: true,
+                    readOnlyHint: false,
+                  },
+                  description: "Mark one item done.",
+                  inputSchema: z.object({ itemId: z.string() }),
+                  outputSchema: pluginToolOutputSchema,
+                  execute: markDone,
+                }),
               }),
-            }),
-          },
-        }),
-      ]),
-    });
-    const { conversationStore, fixture } = await prepareParentConversation();
-    const queue = createConversationWorkQueueTestAdapter();
-    const state = getStateAdapter();
-    await state.connect();
-    try {
-      const created = await createAndEnqueueAgentInvocation(
-        {
-          ...invocationInput,
-          idempotencyKey: "resume-1",
-        },
-        {
-          conversationStore,
-          nowMs: 2_000,
-          queue,
-          state,
-        },
-      );
+            },
+          }),
+        ]),
+      });
+      const { conversationStore, fixture } = await prepareParentConversation();
       const agentRunner = createModelAgentRunner(
         createModelStream([
+          ...(watchEvent ? [{ type: "text" as const, text: "Watching" }] : []),
           {
             type: "toolCall",
             name: "upkeep_markDone",
@@ -469,94 +504,140 @@ describe("agent invocation conversation work", () => {
         ]),
       );
       const run = vi.spyOn(agentRunner, "run");
-      const invocationWorker = createAgentInvocationWorker(agentRunner);
-      const route = routeAgentInvocationWork({
-        fallbackWorker: vi.fn(async () => ({ status: "completed" as const })),
-        invocationWorker,
-      });
+      const {
+        queue,
+        run: route,
+        state,
+      } = await createConversationWorkHarness(agentRunner);
+      try {
+        const created = await createAndEnqueueAgentInvocation(
+          {
+            ...invocationInput,
+            idempotencyKey: "resume-1",
+          },
+          { conversationStore, queue, state },
+        );
+        const { childConversationId } = created;
+        // Child results also wake the parent. This test runs only the child.
+        const processChild = async (softYieldAfterMs?: number) =>
+          await processConversationQueueMessage(
+            queue.takeMessageFor(childConversationId),
+            {
+              conversationStore,
+              queue,
+              run: route,
+              state,
+              ...(softYieldAfterMs === undefined
+                ? undefined
+                : { softYieldAfterMs }),
+            },
+          );
+        if (watchEvent) {
+          // Finish the parent task, then set a timer on the child. The timer
+          // event waits for its burst window, then starts an agent invocation
+          // with the event Source.
+          await processChild();
+          await execute(
+            createEventTools(
+              { ...context(), conversationId: childConversationId },
+              getEventCatalog(),
+            ).watchTimer!,
+            { afterMs: 60_000, intent: "Check the PR checks." },
+            "child-timer",
+          );
+          clock.mockReturnValue(clock() + 120_000);
+          await runTimerWatchHeartbeat({ nowMs: clock(), queue });
+          await processChild();
+          clock.mockReturnValue(clock() + 120_000);
+          await processChild(0);
+        }
 
-      await expect(
-        processConversationQueueMessage(queue.takeMessage(), {
-          conversationStore,
-          queue,
-          run: route,
-          softYieldAfterMs: 0,
-          state,
-        }),
-      ).resolves.toMatchObject({ status: "yielded" });
-      await expect(
-        getAgentInvocation(created.invocationId),
-      ).resolves.toMatchObject({ status: "paused" });
-      await expect(
-        getTurnRecord(
-          created.childConversationId,
-          getAgentInvocationTurnId(created.invocationId),
-        ),
-      ).resolves.toMatchObject({
-        state: "paused",
-        resumeReason: "yield",
-        piMessages: expect.arrayContaining([
-          expect.objectContaining({
-            role: "toolResult",
-            toolName: "upkeep_markDone",
-          }),
-        ]),
-      });
-      expect(markDone).toHaveBeenCalledOnce();
-      expect(markDone.mock.calls[0]?.[0]).toEqual({ itemId: "item-7" });
+        await expect(processChild(0)).resolves.toMatchObject({
+          status: "yielded",
+        });
+        const { invocationId } =
+          (await getActiveAgentInvocationForConversation(childConversationId))!;
+        await expect(
+          getTurnRecord(
+            childConversationId,
+            getAgentInvocationTurnId(invocationId),
+          ),
+        ).resolves.toMatchObject({
+          state: "paused",
+          resumeReason: "yield",
+          source: { kind: watchEvent ? "event" : "agent_invocation" },
+          piMessages: expect.arrayContaining([
+            expect.objectContaining({
+              role: "toolResult",
+              toolName: "upkeep_markDone",
+            }),
+          ]),
+        });
+        if (endBeforeWake) {
+          await completeAgentInvocation({
+            errorMessage: "Agent invocation ended",
+            invocationId,
+            status: "failed",
+          });
+        }
+        if (stopBeforeWake) {
+          // The stop request outlives the paused run. The stopped agent
+          // invocation takes the resume wake and does not run again.
+          await createStopAgentTool(parentConversationId).execute!(
+            { conversation_id: childConversationId },
+            {},
+          );
+        }
+        if (failedResumes > 0) {
+          run.mockRejectedValue(new Error("resume unavailable"));
+        }
+        for (let retry = 0; retry < failedResumes; retry++) {
+          await expect(processChild()).resolves.toMatchObject({
+            status: "failed",
+          });
+          // Each retry wake has a new idempotency key from the clock.
+          clock.mockReturnValue(clock() + 1_000);
+        }
 
-      await expect(
-        processConversationQueueMessage(queue.takeMessage(), {
-          conversationStore,
-          queue,
-          run: route,
-          state,
-        }),
-      ).resolves.toMatchObject({ status: "completed" });
-      expect(run).toHaveBeenCalledTimes(2);
-      await expect(
-        getAgentInvocation(created.invocationId),
-      ).resolves.toMatchObject({
-        result: "Resumed child result",
-        status: "completed",
-      });
-      const projection = await loadProjection({
-        conversationId: created.childConversationId,
-      });
-      expect(
-        projection.filter(
-          (message) =>
-            message.role === "toolResult" &&
-            message.toolName === "upkeep_markDone",
-        ),
-      ).toHaveLength(1);
-      expect(markDone).toHaveBeenCalledOnce();
-      // Guardian records each decision before the action runs. It made none.
-      const events = await getConversationEventStore().loadHistory(
-        created.childConversationId,
-      );
-      expect(
-        events.filter(
-          (event) => event.data.type === "guardian_action_reviewed",
-        ),
-      ).toEqual([]);
-      expect(projection).toEqual(
-        expect.arrayContaining([
-          expect.objectContaining({
-            role: "assistant",
-            content: expect.arrayContaining([
-              expect.objectContaining({
-                type: "text",
-                text: "Resumed child result",
-              }),
-            ]),
-          }),
-        ]),
-      );
-    } finally {
-      await fixture.close();
-    }
-  });
+        await expect(processChild()).resolves.toMatchObject({
+          status: "completed",
+        });
+        expect(run).toHaveBeenCalledTimes(runs);
+        await expect(getAgentInvocation(invocationId)).resolves.toMatchObject(
+          expected,
+        );
+        // An ended child leaves no paused Turn that a later wake could resume.
+        await expect(
+          getTurnRecord(
+            childConversationId,
+            getAgentInvocationTurnId(invocationId),
+          ),
+        ).resolves.not.toMatchObject({ state: "paused" });
+        const projection = await loadProjection({
+          conversationId: childConversationId,
+        });
+        expect(
+          projection.filter(
+            (message) =>
+              message.role === "toolResult" &&
+              message.toolName === "upkeep_markDone",
+          ),
+        ).toHaveLength(1);
+        // The approve tool skips action review, also in a child, and runs
+        // once across the pause and the pick-up.
+        expect(markDone).toHaveBeenCalledOnce();
+        const events =
+          await getConversationEventStore().loadHistory(childConversationId);
+        expect(
+          events.filter(
+            (event) => event.data.type === "guardian_action_reviewed",
+          ),
+        ).toEqual([]);
+      } finally {
+        await fixture.close();
+      }
+    },
+  );
 
   const missing = "Write access to getsentry/junior. Ask the user.";
   it.each([
@@ -601,9 +682,11 @@ describe("agent invocation conversation work", () => {
     "recovers a completed child turn that ends with $name through production routing",
     async ({ expected, tail, turnOutcome }) => {
       const { conversationStore, fixture } = await prepareParentConversation();
-      const queue = createConversationWorkQueueTestAdapter();
-      const state = getStateAdapter();
-      await state.connect();
+      const {
+        queue,
+        run: route,
+        state,
+      } = await createConversationWorkHarness(neverRunAgentRunner());
       try {
         const created = await createAndEnqueueAgentInvocation(
           {
@@ -649,11 +732,6 @@ describe("agent invocation conversation work", () => {
           state: "completed",
           surface: "internal",
         });
-        const route = routeAgentInvocationWork({
-          fallbackWorker: vi.fn(async () => ({ status: "completed" as const })),
-          invocationWorker: createAgentInvocationWorker(neverRunAgentRunner()),
-        });
-
         await expect(
           processConversationQueueMessage(queue.takeMessage(), {
             conversationStore,
diff --git a/scripts/test-architecture-baseline.json b/scripts/test-architecture-baseline.json
index bd310b6fc0..ecec90d703 100644
--- a/scripts/test-architecture-baseline.json
+++ b/scripts/test-architecture-baseline.json
@@ -163,16 +163,15 @@
     "packages/junior/tests/component/task-execution/conversation-work.test.ts": 2,
     "packages/junior/tests/component/task-execution/slack-conversation-work.test.ts": 1,
     "packages/junior/tests/fixtures/acp-http.ts": 1,
-    "packages/junior/tests/fixtures/agent-dispatch.ts": 1,
     "packages/junior/tests/fixtures/chat-runtime.ts": 1,
-    "packages/junior/tests/fixtures/conversation-work.ts": 3,
+    "packages/junior/tests/fixtures/conversation-work.ts": 2,
     "packages/junior/tests/fixtures/conversation.ts": 2,
     "packages/junior/tests/fixtures/oauth-work.ts": 2,
     "packages/junior/tests/integration/acp-http-recovery.test.ts": 2,
     "packages/junior/tests/integration/agent-dispatch-recovery.test.ts": 2,
     "packages/junior/tests/integration/agent-dispatch-work.test.ts": 7,
-    "packages/junior/tests/integration/agent-invocation-concurrency.test.ts": 2,
-    "packages/junior/tests/integration/agent-invocation-work.test.ts": 6,
+    "packages/junior/tests/integration/agent-invocation-concurrency.test.ts": 1,
+    "packages/junior/tests/integration/agent-invocation-work.test.ts": 5,
     "packages/junior/tests/integration/conversation-turn-work.test.ts": 9,
     "packages/junior/tests/integration/durable-queue.test.ts": 2,
     "packages/junior/tests/integration/event-wake-delay.test.ts": 1,