fix(agent-invocations): Route subagent resumes to subagent code

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
 

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

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

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

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

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

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, {

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 {