Unblock fleet waits on interrupt and scope them to the caller

docs/ARCHITECTURE.md

@@ -195,11 +195,11 @@ Invocation: workflows are **not** top-level slash commands. Recipe definitions l
 
 Three distinct concepts (do not conflate them):
 
-| Concept       | What it is                                               | Surface                                                             |
-| ------------- | -------------------------------------------------------- | ------------------------------------------------------------------- |
-| **Agent**     | A runtime entity with its own loop, tools, and context   | Primary session or a spawned child                                  |
-| **Task**      | A checklist item owned by _one_ agent via `manage_tasks` | Local work plan — not a spawn                                       |
-| **Sub-agent** | A short-lived child agent for one self-contained job     | Spawned with the **`task`** tool (wire name kept for compatibility) |
+| Concept       | What it is                                               | Surface                                                   |
+| ------------- | -------------------------------------------------------- | --------------------------------------------------------- |
+| **Agent**     | A runtime entity with its own loop, tools, and context   | Primary session or a spawned child                        |
+| **Task**      | A checklist item owned by _one_ agent via `manage_tasks` | Local work plan — not a spawn                             |
+| **Sub-agent** | A short-lived child agent for one self-contained job     | Spawned with **`spawn_agent`** (or deprecated **`task`**) |
 
 The **`task`** tool **spawns a sub-agent** on a separate inference source (tier/profile resolved from settings). The dispatch brief separates durable `context`, actionable `prompt`, and optional `goals` (checklist seeds for the _child's_ own `manage_tasks` list). The child returns a structured report (`Summary` / `Findings` / `Blockers` / `Paths`) plus a tools-used footer. Parent and child never share a `manage_tasks` list.
 
@@ -219,9 +219,9 @@ 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.** `task-tool.ts` 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 `task` / `search_agents`, 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 the mount always throws `FleetAuthorityError` for a profile-sourced orchestrator (CL-6942/CL-6944 can add one when a real caller needs it). `FLEET_VERBS` in `authority.ts` also names the not-yet-implemented verbs (`spawn_agent`, `wait_agents`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `read_agent_trace`, `followup_task`) so their future mount sites inherit the same gate.
-- **Subtree authority — a seam, not yet wired.** `assertCanTargetAgent(actor, targetId, nodes)` (`src/subagent/authority.ts`) implements the "root owns its tree; a child manages only its own descendants" rule (Tier 1 may target anyone, Tier 2 may target only its own descendants over the same `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks, Tier 3 always fails closed) — but **it has no production call site yet**. No verb today lets one live agent address another (`task` only spawns), so this rule is exercised only by `authority.test.ts` and is not enforced at runtime in this PR. It exists so CL-6942 (split spawn from wait) and CL-6944 (`send_input` steering) — the first verbs that make an agent addressable by another — can call it from day one instead of each inventing its own check. Treat it as unenforced until one of those wires a call site.
-- `task()` is unaffected and remains the only spawn verb until the new verbs land beside it (deprecated-not-deleted per the CL-6940 epic). Its argument schema and wire contract are unchanged; the tier check only gates which packages may have it mounted at all.
+- **Mount-time gate — live today, and fails closed.** `task-tool.ts` 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 the mount always throws `FleetAuthorityError` for a profile-sourced orchestrator. `FLEET_VERBS` in `authority.ts` names the live verbs (`task`, `spawn_agent`, `wait_agents`, `interrupt_agent`, `close_agent`, `resume_agent`, `followup_task`, `read_agent_trace`, `search_agents`) plus reserved names (`list_agents`, `send_input`) so a later mount site inherits the same gate.
+- **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. `read_agent_trace` is a production call site. `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 own `fleetRecords`, not every running session in the shared store. `interrupt_agent` terminalizes the wait mailbox immediately.
+- `task()` remains the deprecated fused spawn+wait fallback. `spawn_agent` + `wait_agents` is the supported parallel path. The tier check still gates which packages may mount any fleet verb.
 
 #### Closed director fleet (`src/agent/directors/`)
 

src/agent/directors/skywalker/package.test.ts

@@ -102,6 +102,8 @@ describe("skywalkerPackage", () => {
     expect(p).toContain("wait_agents");
     expect(p).toContain("Idle-orchestrator");
     expect(p).toContain("deprecated fused spawn+wait");
+    expect(p).toContain('mode="all"');
+    expect(p).toContain("uncollected spawns");
     expect(p).not.toContain("Present the plan when the change is large or ambiguous");
   });
 

src/agent/directors/skywalker/package.ts

@@ -16,7 +16,7 @@ You do not do the specialists' jobs by default. For tiny bounded product edits,
 
 Do not run long-blocking jobs on the parent (evals, full test suites, long installs, long-running implementation). Dispatch intern (mechanical shell), tester (suite / repro), or builder (substantial code). Path tools (write_file/edit_file/delete_file) are the DIY surface; shell file-writes stay denied.
 
-Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents / task() right after spawn. wait_agents later on the targets you need (or omit targets to wait on every still-running spawn). task() still fuses spawn+wait and holds the parent until that one worker finishes. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents / task() holds those steers. A bare spawn_agent does not.
+Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents / task() right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. task() still fuses spawn+wait and holds the parent until that one worker finishes. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents / task() holds those steers. A bare spawn_agent does not.
 
 # Operator updates (mandatory while fleet is live)
 

src/agent/tools.ts

@@ -346,7 +346,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
         createWaitAgentsTool({ sessions: fleetSessions, fleetRecords }),
         createCloseAgentTool({ sessions: fleetSessions }),
         createResumeAgentTool({ sessions: fleetSessions }),
-        createInterruptAgentTool({ sessions: fleetSessions }),
+        createInterruptAgentTool({ sessions: fleetSessions, fleetRecords }),
         createFollowupTaskTool({ sessions: fleetSessions }),
       );
     }

src/subagent/agent-fleet.test.ts

@@ -7,6 +7,7 @@ import {
   MAX_FLEET_RECORDS,
   type AgentFleetDeps,
 } from "./agent-fleet.js";
+import { createInterruptAgentTool } from "./lifecycle-tools.js";
 import { createSubAgentSessionStore } from "./session-store.js";
 import { createPermissionGate } from "../permission/gate.js";
 import { forcedStopReport } from "./stop-policy.js";
@@ -173,7 +174,7 @@ describe("spawn_agent + wait_agents", () => {
     expect(secondResults[0]!.report).toBe("finished");
   });
 
-  test("wait_agents with no targets waits on all currently running spawned agents", async () => {
+  test("wait_agents with no targets waits on all uncollected agents in this fleet", async () => {
     const gates = [deferred<RunSubAgentResult>(), deferred<RunSubAgentResult>()];
     let callIndex = 0;
     const deps = makeDeps(async () => gates[callIndex++]!.promise);
@@ -374,3 +375,258 @@ describe("fleetRecords retention cap", () => {
     expect(results[0]!.hint).toContain("read_agent_trace");
   });
 });
+
+describe("spawn_agent parentage", () => {
+  test("records the caller session as parentSessionId", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const deps = makeDeps(async () => gate.promise);
+    deps.parentSessionId = "parent-orch";
+    const spawn = createSpawnAgentTool(deps);
+
+    const spawned = await callTool(spawn, {
+      description: "child",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const session = deps.sessions.get(spawned.agent_id as string);
+    expect(session?.parentSessionId).toBe("parent-orch");
+
+    gate.resolve({ report: "done" });
+  });
+});
+
+describe("wait_agents caller scope", () => {
+  test("omitted targets wait only on this fleet, not every running session in the shared store", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const deps = makeDeps(async () => gate.promise);
+    const foreign = deps.sessions.start({
+      id: "foreign-sibling",
+      description: "someone else's worker",
+      agentId: "explorer",
+      brief: "b",
+    });
+    deps.sessions.markRunning(foreign.id);
+
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const spawned = await callTool(spawn, {
+      description: "mine",
+      prompt: "do it",
+      intent: "explore",
+    });
+
+    const waited = await callTool(wait, { timeout_ms: 50 });
+    expect(waited.timed_out).toBe(true);
+    const results = waited.results as { agent_id: string; status: string }[];
+    expect(results.map((r) => r.agent_id)).toEqual([spawned.agent_id as string]);
+    expect(results.every((r) => r.agent_id !== foreign.id)).toBe(true);
+
+    gate.resolve({ report: "done" });
+  });
+
+  test("mode=all stays blocked until every target is terminal", async () => {
+    const gates = [deferred<RunSubAgentResult>(), deferred<RunSubAgentResult>()];
+    let callIndex = 0;
+    const deps = makeDeps(async () => gates[callIndex++]!.promise);
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+
+    const first = await callTool(spawn, {
+      description: "a",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const second = await callTool(spawn, {
+      description: "b",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const ids = [first.agent_id as string, second.agent_id as string];
+
+    gates[0]!.resolve({ report: "a done" });
+    const partial = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 50 });
+    expect(partial.timed_out).toBe(true);
+    const partialResults = partial.results as { status: string }[];
+    expect(partialResults.some((r) => r.status === "running")).toBe(true);
+
+    gates[1]!.resolve({ report: "b done" });
+    const finished = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 5000 });
+    expect(finished.timed_out).toBe(false);
+    const finishedResults = finished.results as { status: string }[];
+    expect(finishedResults.every((r) => r.status === "done")).toBe(true);
+  });
+
+  test("mode=all with one interrupted target stays blocked until siblings finish", async () => {
+    const gates = [deferred<RunSubAgentResult>(), deferred<RunSubAgentResult>()];
+    let callIndex = 0;
+    const deps = makeDeps(async (params) => {
+      params.onAgentReady?.({
+        close: async () => {},
+        interrupt: () => {},
+        followup: async () => "",
+      });
+      return gates[callIndex++]!.promise;
+    });
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const interrupt = createInterruptAgentTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+
+    const first = await callTool(spawn, {
+      description: "a",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const second = await callTool(spawn, {
+      description: "b",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const ids = [first.agent_id as string, second.agent_id as string];
+
+    // Interrupt one of N before mode=all starts: interrupted is terminal for
+    // that target, but mode=all must not complete as "all done" while a
+    // sibling is still running.
+    if (interrupt.kind !== "full") throw new Error("expected full tool");
+    await interrupt.handler(
+      { id: "int-1", name: "interrupt_agent", arguments: { target: ids[0]! } },
+      new AbortController().signal,
+    );
+
+    const partial = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 50 });
+    expect(partial.timed_out).toBe(true);
+    const partialResults = partial.results as { agent_id: string; status: string }[];
+    expect(partialResults.find((r) => r.agent_id === ids[0]!)?.status).toBe("interrupted");
+    expect(partialResults.find((r) => r.agent_id === ids[1]!)?.status).toBe("running");
+
+    gates[1]!.resolve({ report: "b done" });
+    const finished = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 5000 });
+    expect(finished.timed_out).toBe(false);
+    const finishedResults = finished.results as { agent_id: string; status: string }[];
+    expect(finishedResults.find((r) => r.agent_id === ids[0]!)?.status).toBe("interrupted");
+    expect(finishedResults.find((r) => r.agent_id === ids[1]!)?.status).toBe("done");
+    // Leave the interrupted gate unresolved — interrupt unblocked the wait
+    // without the run settling.
+  });
+
+  test("aborting the wait returns without cancelling workers", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const deps = makeDeps(async () => gate.promise);
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const spawned = await callTool(spawn, {
+      description: "slow",
+      prompt: "do it",
+      intent: "explore",
+    });
+    const id = spawned.agent_id as string;
+
+    if (wait.kind !== "full") throw new Error("expected full tool");
+    const ac = new AbortController();
+    const started = Date.now();
+    const pending = wait.handler(
+      { id: "wait-1", name: "wait_agents", arguments: { targets: [id], timeout_ms: 5000 } },
+      ac.signal,
+    );
+    ac.abort();
+    const result = await pending;
+    expect(Date.now() - started).toBeLessThan(500);
+    const content =
+      typeof result.content === "string" ? result.content : JSON.stringify(result.content);
+    const parsed = JSON.parse(content) as {
+      timed_out: boolean;
+      results: { status: string }[];
+    };
+    expect(parsed.timed_out).toBe(true);
+    expect(parsed.results[0]!.status).toBe("running");
+    expect(deps.sessions.get(id)?.status).toBe("running");
+
+    gate.resolve({ report: "done" });
+  });
+});
+
+describe("interrupt_agent unblocks wait_agents", () => {
+  test("interrupt marks the fleet record terminal so wait returns without the run settling", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const deps = makeDeps(async (params) => {
+      params.onAgentReady?.({
+        close: async () => {},
+        interrupt: () => {},
+        followup: async () => "",
+      });
+      return gate.promise;
+    });
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      sessions: deps.sessions,
+      fleetRecords: deps.fleetRecords,
+    });
+    const interrupt = createInterruptAgentTool({
+      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 });
+    if (interrupt.kind !== "full") throw new Error("expected full tool");
+    await interrupt.handler(
+      { id: "int-1", name: "interrupt_agent", arguments: { target: id } },
+      new AbortController().signal,
+    );
+
+    const waited = await waiting;
+    expect(waited.timed_out).toBe(false);
+    const results = waited.results as { agent_id: string; status: string }[];
+    expect(results).toEqual([{ agent_id: id, status: "interrupted" }]);
+    expect(deps.sessions.get(id)?.lifecycleStatus).toBe("interrupted");
+    expect(deps.sessions.get(id)?.status).toBe("running");
+  });
+
+  test("an interrupted run result terminalizes a still-running fleet record", async () => {
+    const settle = deferred<RunSubAgentResult>();
+    const deps = makeDeps(async () => settle.promise);
+    const spawn = createSpawnAgentTool(deps);
+    const wait = createWaitAgentsTool({
+      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;
+
+    settle.resolve({
+      report: "## Summary\nStopped.\n## Findings\npartial\n## Blockers\ninterrupted\n## Paths\n",
+      interrupted: true,
+    });
+
+    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("partial");
+  });
+});