Merge pull request #858 from corbitsdev/cl-7331-interrupted-workers-never-return-queued-follow-up-reports

Keep wait live after interrupt-with-follow-up

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -233,7 +233,7 @@ Every director package carries a required `tier: SubagentTier` field (`src/agent
 Enforcement is runtime code at the existing tool-mount point, not prompt wording — this is the fix for four prior mechanisms (`writePaths`, `report.requiredSections`, a `--config` comment, the thrash matcher) that were documented-as-enforced while enforcing nothing:
 
 - **Mount-time gate — live today, and fails closed.** `spawn_agent` resolves the caller's tier at dispatch time — a closed director's `DirectorPackage.tier` — and forwards it as `RunSubAgentParams.orchestratorTier`. `runSubAgent` (`src/subagent/run.ts`) then calls `assertTierMayMountFleetVerb(tier, toolName)` (`src/subagent/authority.ts`) before installing fleet verbs, treating a **missing** `orchestratorTier` as `"leaf"` — deny, not skip. This is the case that matters most: a project-local or plugin `AgentProfile` with `orchestrator: true` is outside the closed director set and is **not** trusted with fleet verbs just because `orchestrator: true` is set — there is no profile-level opt-in today, so `spawn_agent` rejects a profile-sourced orchestrator before starting a session. `FLEET_VERBS` in `authority.ts` names the live verbs (`spawn_agent`, `wait_agents`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `read_agent_trace`, `search_agents`) so every mount site inherits the same gate. `list_agents` is the non-blocking mailbox-scoped list of this install's own `spawn_agent` workers (same scope as `wait_agents`); nested orchestrators may mount it. Fleet discovery (`search_agents`) remains Tier 1 only.
-- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `wait_agents` explicit targets, `send_input`, `interrupt_agent`, `close_agent`, and `resume_agent`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's per-install wait mailbox over the shared session store, not every running session. Wait JSON is a projection of stored lifecycle plus mailbox membership/pin/collected/interrupt override — not a second terminal store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` stamps the session `interrupted`; wait JSON projects that stored lifecycle and does not write a mailbox overlay. `send_input` with `interrupt:true` sets the mailbox interrupt overlay so wait unblocks while a queued followup may already be running. The wait path collects a terminal status so a later followup cannot resurrect an already-observed interrupt. `close_agent` terminalizes the wait mailbox before teardown. Operator cancel (`cancel` / `cancelAll`) projects wait status `interrupted`.
+- **Subtree authority — wired for addressing verbs.** `assertCanTargetAgent(actor, targetId, nodes)` implements the "root owns its tree; a child manages only its own descendants" rule over the `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks. Production call sites: `read_agent_trace`, `wait_agents` explicit targets, `send_input`, `interrupt_agent`, `close_agent`, and `resume_agent`. Nested mounts pass `{actorId, tier, getNodes}` from `run.ts`; a missing `actorId` fails closed. Tier-1 primary omits authority and stays unrestricted. `spawn_agent` records `parentSessionId` on nested workers so `close_agent`'s descendant walk can see them. `wait_agents` with omitted targets waits only on that caller's per-install wait mailbox over the shared session store, not every running session. Wait JSON is a projection of stored lifecycle plus mailbox membership/pin/collected/interrupt override — not a second terminal store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` stamps the session `interrupted` and, with `close_agent`, writes the mailbox interrupt overlay; `send_input` with `interrupt:true` does not. `send_input` with `interrupt:true` leaves wait live (`running`/`queued`) until the followup settles. `close_agent` terminalizes the wait mailbox before teardown. Operator cancel (`cancel` / `cancelAll`) projects wait status `interrupted`.
 - **Leaf `ask_director` (CL-6945).** Tier 3 leaves mount `ask_director` (not `ask_operator`). The worker evaluates caps, then awaits a session-store port; a missing port returns an error and does not suspend. `wait_agents` projects `awaiting_director` with `question` / `question_id` / `description` — this is wait JSON only, not a `WorkerLifecycle` state. Re-wait while still pending re-delivers the same question. Soft `send_input` answers the pending ask (it does not deliver a steer inbound). Interrupt / settle / close cancel the ask, including descendants. A worker blocked in `ask_director` is not stall-salvaged.
 - `spawn_agent` + `wait_agents` is the only spawn path. The tier check still gates which packages may mount any fleet verb.
 

src/subagent/agent-fleet.test.ts

@@ -16,6 +16,7 @@ import { isLiveWaitStatus } from "./lifecycle.js";
 import {
   createInterruptAgentTool,
   createCloseAgentTool,
+  createResumeAgentTool,
   createSendInputTool,
 } from "./lifecycle-tools.js";
 import { createSubAgentSessionStore } from "./session-store.js";
@@ -1072,7 +1073,7 @@ describe("interrupt_agent unblocks wait_agents", () => {
     gate.resolve({ report: "done" });
   });
 
-  test("send_input interrupt:true unblocks wait_agents as interrupted", async () => {
+  test("send_input interrupt:true keeps wait_agents live until the followup completes", async () => {
     const gate = deferred<RunSubAgentResult>();
     const followupGate = deferred<string>();
     const deps = makeDeps(async (params) => {
@@ -1101,15 +1102,351 @@ describe("interrupt_agent unblocks wait_agents", () => {
     const id = spawned.agent_id as string;
     const waiting = callTool(wait, { targets: [id], timeout_ms: 5000 });
     await callTool(sendInput, { target: id, message: "stop that", interrupt: true });
+    followupGate.resolve("later");
+    gate.resolve({ report: "original interrupted", interrupted: true } as RunSubAgentResult);
     const waited = await waiting;
     expect(waited.timed_out).toBe(false);
     const results = waited.results as {
       agent_id: string;
       status: string;
+      report?: string;
       stop_reason?: string;
     }[];
-    expect(results).toEqual([{ agent_id: id, status: "interrupted", stop_reason: "interrupted" }]);
-    expect(deps.sessions.get(id)?.stopReason).toBe("interrupted");
+    expect(results[0]!.status).toBe("done");
+    expect(results[0]!.report).toBe("later");
+  });
+
+  test("CL-7331: send_input interrupt keeps wait live until the queued followup completes", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const followupGate = deferred<string>();
+    const deps = makeDeps(async (params) => {
+      params.onAgentReady?.({
+        close: async () => {},
+        interrupt: () => {},
+        followup: async () => followupGate.promise,
+        deliver: () => {},
+      });
+      return gate.promise;
+    });
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const list = createListAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const sendInput = createSendInputTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const resume = createResumeAgentTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const spawned = await callTool(spawn, {
+      description: "looping",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const id = spawned.agent_id as string;
+
+    const sent = await callTool(sendInput, {
+      target: id,
+      message: "return a concise report",
+      interrupt: true,
+    });
+    expect(sent.status).toBe("interrupted");
+
+    // The queued followup is still running: wait must stay live (not an
+    // immediate terminal interrupted), and list must agree with lifecycle.
+    const pending = await callTool(wait, { targets: [id], timeout_ms: 50 });
+    expect(pending.timed_out).toBe(true);
+    expect((pending.results as { status: string }[])[0]!.status).toBe("running");
+
+    const listed = await callTool(list, {});
+    const entry = (listed.agents as { agent_id: string; status: string; lifecycle: string }[]).find(
+      (a) => a.agent_id === id,
+    );
+    expect(entry?.status).toBe("running");
+    expect(entry?.lifecycle).toBe("running");
+
+    // A resume while the followup is in flight must agree with wait/list.
+    if (resume.kind !== "full") throw new Error("expected full tool");
+    const resumed = await resume.handler(
+      {
+        id: "resume-while-followup",
+        name: "resume_agent",
+        arguments: { target: id, message: "x" },
+      },
+      new AbortController().signal,
+    );
+    expect(resumed.isError).toBe(true);
+    expect(String(resumed.content)).toContain("status: running");
+
+    // When the queued followup finishes, its report must surface via wait.
+    followupGate.resolve("followup report");
+    gate.resolve({ report: "original interrupted", interrupted: true } as RunSubAgentResult);
+    const done = await callTool(wait, { targets: [id], timeout_ms: 5000 });
+    expect(done.timed_out).toBe(false);
+    const doneResults = done.results as { status: string; report?: string }[];
+    expect(doneResults[0]!.status).toBe("done");
+    expect(doneResults[0]!.report).toBe("followup report");
+  });
+
+  test("close_agent overlay survives a send_input followup completing in the close window", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const followupGate = deferred<string>();
+    const closeHold = deferred<undefined>();
+    const deps = makeDeps(async (params) => {
+      params.onAgentReady?.({
+        close: async () => closeHold.promise,
+        interrupt: () => {},
+        followup: async () => followupGate.promise,
+        deliver: () => {},
+      });
+      return gate.promise;
+    });
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const sendInput = createSendInputTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const close = createCloseAgentTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const spawned = await callTool(spawn, {
+      description: "looping",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const id = spawned.agent_id as string;
+
+    const waiting = callTool(wait, { targets: [id], timeout_ms: 5000 });
+    await callTool(sendInput, { target: id, message: "stop that", interrupt: true });
+    if (close.kind !== "full") throw new Error("expected full tool");
+    const closing = close.handler(
+      { id: "close-during-followup", name: "close_agent", arguments: { target: id } },
+      new AbortController().signal,
+    );
+    followupGate.resolve("followup during close");
+    gate.resolve({ report: "original interrupted", interrupted: true } as RunSubAgentResult);
+
+    const waited = await waiting;
+    expect(waited.timed_out).toBe(false);
+    const results = waited.results as { status: string }[];
+    expect(results[0]!.status).toBe("interrupted");
+    expect(deps.fleetRecords.peek(id)?.status).toBe("interrupted");
+
+    closeHold.resolve(undefined);
+    await closing;
+  });
+
+  test("close overlay without in-flight wait stays interrupted after followup complete", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const followupGate = deferred<string>();
+    const closeHold = deferred<undefined>();
+    const deps = makeDeps(async (params) => {
+      params.onAgentReady?.({
+        close: async () => closeHold.promise,
+        interrupt: () => {},
+        followup: async () => followupGate.promise,
+        deliver: () => {},
+      });
+      return gate.promise;
+    });
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const list = createListAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const sendInput = createSendInputTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const close = createCloseAgentTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const spawned = await callTool(spawn, {
+      description: "looping",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const id = spawned.agent_id as string;
+
+    await callTool(sendInput, { target: id, message: "stop that", interrupt: true });
+    if (close.kind !== "full") throw new Error("expected full tool");
+    const closing = close.handler(
+      { id: "close-held-then-followup", name: "close_agent", arguments: { target: id } },
+      new AbortController().signal,
+    );
+
+    followupGate.resolve("followup after close overlay");
+    await new Promise<void>((resolve) => {
+      const done = (): boolean => deps.sessions.get(id)?.lifecycle.state === "completed";
+      if (done()) {
+        resolve();
+        return;
+      }
+      const unsub = deps.sessions.subscribe(() => {
+        if (done()) {
+          unsub();
+          resolve();
+        }
+      });
+      if (done()) {
+        unsub();
+        resolve();
+      }
+    });
+
+    expect(deps.fleetRecords.peek(id)?.status).toBe("interrupted");
+    const listed = await callTool(list, {});
+    const entry = (listed.agents as { agent_id: string; status: string }[]).find(
+      (a) => a.agent_id === id,
+    );
+    expect(entry?.status).toBe("interrupted");
+
+    const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 });
+    expect(waited.timed_out).toBe(false);
+    const results = waited.results as { status: string }[];
+    expect(results[0]!.status).toBe("interrupted");
+
+    closeHold.resolve(undefined);
+    await closing;
+    gate.resolve({ report: "original interrupted", interrupted: true } as RunSubAgentResult);
+  });
+
+  test("completeAfterInterrupt does not clear a close overlay", () => {
+    const sessions = createSubAgentSessionStore();
+    const fleetRecords = createFleetMailbox(sessions);
+    const worker = sessions.start({
+      description: "looping",
+      agentId: "explorer",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(worker.id);
+    fleetRecords.register(worker.id);
+    fleetRecords.noteFollowup(worker.id);
+    fleetRecords.interrupt(worker.id);
+    expect(fleetRecords.peek(worker.id)?.status).toBe("interrupted");
+    fleetRecords.completeAfterInterrupt(worker.id, "followup reply");
+    expect(fleetRecords.peek(worker.id)?.status).toBe("interrupted");
+  });
+
+  test("rejected send_input followup clears the lane so wait collects interrupted salvage", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const followupGate = deferred<string>();
+    const deps = makeDeps(async (params) => {
+      params.onAgentReady?.({
+        close: async () => {},
+        interrupt: () => {},
+        followup: async () => followupGate.promise,
+        deliver: () => {},
+      });
+      return gate.promise;
+    });
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const sendInput = createSendInputTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const spawned = await callTool(spawn, {
+      description: "looping",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const id = spawned.agent_id as string;
+    await callTool(sendInput, { target: id, message: "stop that", interrupt: true });
+    followupGate.reject(new Error("followup failed"));
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    gate.resolve({
+      report: "## Summary\nStopped.\n## Findings\nsalvage\n## Blockers\ninterrupted\n## Paths\n",
+      interrupted: true,
+    } as RunSubAgentResult);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 });
+    expect(waited.timed_out).toBe(false);
+    const results = waited.results as { status: string; report?: string }[];
+    expect(results[0]!.status).toBe("interrupted");
+    expect(results[0]!.report).toContain("salvage");
+  });
+
+  test("send_input interrupt queued overlay clears when the followup is admitted", async () => {
+    const admission = createAdmissionQueue({ capacity: 1 });
+    const sessions = createSubAgentSessionStore({ admission });
+    const fleetRecords = createFleetMailbox(sessions);
+    admission.enqueue({
+      id: "holder",
+      provider: "p",
+      start: () => {},
+    });
+    const worker = sessions.start({
+      description: "looping",
+      agentId: "explorer",
+      brief: "b",
+      retained: true,
+      provider: "p",
+    });
+    sessions.markRunning(worker.id);
+    fleetRecords.register(worker.id);
+    const followupGate = deferred<string>();
+    sessions.registerInterrupt(worker.id, () => {});
+    sessions.registerFollowup(worker.id, async () => followupGate.promise);
+
+    const sendInput = createSendInputTool({ sessions, fleetRecords });
+    const wait = createWaitAgentsTool({ sessions, fleetRecords });
+    const list = createListAgentsTool({ sessions, fleetRecords });
+
+    const sent = await callTool(sendInput, {
+      target: worker.id,
+      message: "stop that",
+      interrupt: true,
+    });
+    expect(sent.status).toBe("interrupted");
+    expect(sessions.get(worker.id)?.lifecycleStatus).toBe("pending_init");
+
+    const queuedWait = await callTool(wait, { targets: [worker.id], timeout_ms: 50 });
+    expect(queuedWait.timed_out).toBe(true);
+    expect((queuedWait.results as { status: string }[])[0]!.status).toBe("queued");
+    const queuedList = await callTool(list, {});
+    const queuedEntry = (
+      queuedList.agents as { agent_id: string; status: string; lifecycle: string }[]
+    ).find((a) => a.agent_id === worker.id);
+    expect(queuedEntry?.status).toBe("queued");
+    expect(queuedEntry?.lifecycle).toBe("pending_init");
+
+    admission.release("holder");
+    await new Promise((resolve) => setTimeout(resolve, 20));
+
+    expect(sessions.get(worker.id)?.lifecycleStatus).toBe("running");
+    const runningWait = await callTool(wait, { targets: [worker.id], timeout_ms: 50 });
+    expect(runningWait.timed_out).toBe(true);
+    expect((runningWait.results as { status: string }[])[0]!.status).toBe("running");
+    const runningList = await callTool(list, {});
+    const runningEntry = (
+      runningList.agents as { agent_id: string; status: string; lifecycle: string }[]
+    ).find((a) => a.agent_id === worker.id);
+    expect(runningEntry?.status).toBe("running");
+    expect(runningEntry?.lifecycle).toBe("running");
+
     followupGate.resolve("later");
   });