Add send_input without breaking the wait mailbox (#682)

* Add send_input without breaking the wait mailbox

Soft-deliver steers a running worker without completing wait_agents.
interrupt:true uses the same mailbox flip as interrupt_agent so a parent
wait unblocks once, then a later followup can become done only if that
interrupt was never collected.

* Gate addressing fleet verbs with subtree authority

Nested interrupt/close/resume/followup now share send_input's
assertCanTargetAgent check and fail closed without an actorId.
Soft-interrupt wait_agents collects so a later followup cannot
resurrect an already-observed interrupt as done.

* Format run.ts after send_input authority restack

Author: TheGreatAxios

docs/ARCHITECTURE.md

@@ -219,8 +219,8 @@ 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 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`, `list_agents`, `interrupt_agent`, `close_agent`, `resume_agent`, `followup_task`, `read_agent_trace`, `search_agents`) plus reserved name (`send_input`) so a later 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. `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. `list_agents` reports that same mailbox without blocking. `interrupt_agent` and `close_agent` terminalize the wait mailbox immediately.
+- **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`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `followup_task`, `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`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, and `followup_task`. 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 own `fleetRecords`, not every running session in the shared store. `list_agents` reports that same mailbox without blocking. `interrupt_agent` / `send_input` with `interrupt:true` terminalize the wait mailbox immediately; the soft-interrupt wait path collects so a later followup cannot resurrect an already-observed interrupt. `close_agent` also terminalizes the wait mailbox before teardown.
 - `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/fleet-verbs-mount.test.ts

@@ -18,6 +18,7 @@ const FLEET_VERBS = [
   "resume_agent",
   "interrupt_agent",
   "followup_task",
+  "send_input",
 ] as const;
 
 describe("primary fleet verb mount", () => {

src/agent/tool-search.test.ts

@@ -93,6 +93,7 @@ describe("createToolIndex", () => {
       "resume_agent",
       "interrupt_agent",
       "followup_task",
+      "send_input",
     ] as const) {
       expect(CORE_TOOL_NAMES).toContain(name);
       expect(advertised).toContain(name);
@@ -243,6 +244,7 @@ describe("advertisedTools", () => {
       "resume_agent",
       "interrupt_agent",
       "followup_task",
+      "send_input",
     ] as const) {
       expect(prefix).toContain(name);
     }

src/agent/tool-search.ts

@@ -50,6 +50,7 @@ export const CORE_TOOL_NAMES: readonly string[] = [
   "resume_agent",
   "interrupt_agent",
   "followup_task",
+  "send_input",
 ];
 
 const ORCHESTRATOR_ONLY_TOOL_NAMES: readonly string[] = [
@@ -62,6 +63,7 @@ const ORCHESTRATOR_ONLY_TOOL_NAMES: readonly string[] = [
   "resume_agent",
   "interrupt_agent",
   "followup_task",
+  "send_input",
 ];
 
 // Session-start facts that gate a core tool's advertisement. Each must be

src/agent/tools.ts

@@ -50,6 +50,7 @@ import {
   createResumeAgentTool,
   createInterruptAgentTool,
   createFollowupTaskTool,
+  createSendInputTool,
 } from "../subagent/lifecycle-tools.js";
 import { parseManageTasksArgs } from "./tasks.js";
 import { createListDirTool } from "../util/list-dir.js";
@@ -350,6 +351,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
         createResumeAgentTool({ sessions: fleetSessions }),
         createInterruptAgentTool({ sessions: fleetSessions, fleetRecords }),
         createFollowupTaskTool({ sessions: fleetSessions }),
+        createSendInputTool({ sessions: fleetSessions, fleetRecords }),
       );
     }
   }

src/subagent/agent-fleet.test.ts

@@ -8,7 +8,11 @@ import {
   MAX_FLEET_RECORDS,
   type AgentFleetDeps,
 } from "./agent-fleet.js";
-import { createInterruptAgentTool, createCloseAgentTool } from "./lifecycle-tools.js";
+import {
+  createInterruptAgentTool,
+  createCloseAgentTool,
+  createSendInputTool,
+} from "./lifecycle-tools.js";
 import { createSubAgentSessionStore } from "./session-store.js";
 import { createPermissionGate } from "../permission/gate.js";
 import { forcedStopReport } from "./stop-policy.js";
@@ -471,6 +475,7 @@ describe("wait_agents caller scope", () => {
         close: async () => {},
         interrupt: () => {},
         followup: async () => "",
+        deliver: () => {},
       });
       return gates[callIndex++]!.promise;
     });
@@ -568,6 +573,7 @@ describe("interrupt_agent unblocks wait_agents", () => {
         close: async () => {},
         interrupt: () => {},
         followup: async () => "",
+        deliver: () => {},
       });
       return gate.promise;
     });
@@ -631,13 +637,84 @@ describe("interrupt_agent unblocks wait_agents", () => {
     expect(results[0]!.report).toContain("partial");
   });
 
+  test("send_input soft-deliver does not complete wait_agents", async () => {
+    const gate = deferred<RunSubAgentResult>();
+    const deps = makeDeps(async (params) => {
+      params.onAgentReady?.({
+        close: async () => {},
+        interrupt: () => {},
+        followup: async () => "",
+        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: "keep going" });
+    const waited = await callTool(wait, { targets: [id], timeout_ms: 50 });
+    expect(waited.timed_out).toBe(true);
+    const results = waited.results as { status: string }[];
+    expect(results[0]!.status).toBe("running");
+    gate.resolve({ report: "done" });
+  });
+
+  test("send_input interrupt:true unblocks wait_agents as interrupted", 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;
+    const waiting = callTool(wait, { targets: [id], timeout_ms: 5000 });
+    await callTool(sendInput, { target: id, message: "stop that", interrupt: true });
+    const waited = await waiting;
+    expect(waited.timed_out).toBe(false);
+    const results = waited.results as { status: string }[];
+    expect(results[0]!.status).toBe("interrupted");
+    followupGate.resolve("later");
+  });
+
   test("soft-interrupt wait path collects so omitted re-wait does not re-deliver", async () => {
     const gate = deferred<RunSubAgentResult>();
     const deps = makeDeps(async (params) => {
       params.onAgentReady?.({
         close: async () => {},
         interrupt: () => {},
         followup: async () => "",
+        deliver: () => {},
       });
       return gate.promise;
     });
@@ -678,6 +755,7 @@ describe("interrupt_agent unblocks wait_agents", () => {
         close: async () => {},
         interrupt: () => {},
         followup: async () => "",
+        deliver: () => {},
       });
       return settle.promise;
     });
@@ -722,6 +800,35 @@ describe("interrupt_agent unblocks wait_agents", () => {
     expect(results[0]!.status).toBe("interrupted");
     expect(results[0]!.report).toContain("salvage");
   });
+
+  test("soft-interrupt wait collects so a later followup cannot resurrect done", async () => {
+    const sessions = createSubAgentSessionStore();
+    const fleetRecords = createFleetRecords();
+    const worker = sessions.start({
+      id: "soft-int",
+      description: "looping",
+      agentId: "explorer",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(worker.id);
+    // Running fleet record + soft-interrupted session (lifecycle only) —
+    // the wait soft path must interrupt+take before returning.
+    fleetRecords.register(worker.id);
+    sessions.registerInterrupt(worker.id, () => {});
+    sessions.interruptOne(worker.id);
+
+    const wait = createWaitAgentsTool({ sessions, fleetRecords });
+    const waited = await callTool(wait, { targets: [worker.id], timeout_ms: 1000 });
+    expect(waited.timed_out).toBe(false);
+    const results = waited.results as { status: string }[];
+    expect(results[0]!.status).toBe("interrupted");
+    expect(fleetRecords.peek(worker.id)?.collected).toBe(true);
+
+    fleetRecords.completeAfterInterrupt(worker.id, "resurrected reply");
+    expect(fleetRecords.peek(worker.id)?.status).toBe("interrupted");
+    expect(fleetRecords.peek(worker.id)?.collected).toBe(true);
+  });
 });
 
 describe("close_agent unblocks wait_agents", () => {
@@ -732,6 +839,7 @@ describe("close_agent unblocks wait_agents", () => {
         close: async () => {},
         interrupt: () => {},
         followup: async () => "",
+        deliver: () => {},
       });
       return gate.promise;
     });

src/subagent/agent-fleet.ts

@@ -157,6 +157,20 @@ class FleetRecords {
     this.notify();
   }
 
+  /**
+   * send_input interrupt:true queued a followup that has now finished.
+   * Upgrade an uncollected interrupted record to done. No-op if wait_agents
+   * already collected the interrupt, so a later reply cannot resurrect it.
+   */
+  completeAfterInterrupt(id: string, report: string): void {
+    const existing = this.records.get(id);
+    if (existing === undefined || existing.collected === true) return;
+    if (existing.status !== "interrupted") return;
+    this.records.set(id, { status: "done", report });
+    this.enforceCap();
+    this.notify();
+  }
+
   ids(): string[] {
     return [...this.records.keys()];
   }
@@ -522,10 +536,11 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
         // Keep the session open after a clean completion, and hand the
         // store a bounded close for close_agent to call later.
         persist: true,
-        onAgentReady: ({ close, interrupt, followup }) => {
+        onAgentReady: ({ close, interrupt, followup, deliver }) => {
           deps.sessions.registerClose(session.id, close);
           deps.sessions.registerInterrupt(session.id, interrupt);
           deps.sessions.registerFollowup(session.id, followup);
+          deps.sessions.registerDeliver(session.id, deliver);
           deps.sessions.markRunning(session.id);
         },
       };
@@ -706,8 +721,9 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool {
         }
         const session = deps.sessions.get(id);
         if (isSoftInterrupted(session)) {
-          // Terminalize + collect so an omitted-targets re-wait does not keep
-          // seeing this id as uncollected / re-deliver soft-interrupt.
+          // Match the mailbox to what we report (include salvage report when
+          // present), then collect so a later completeAfterInterrupt cannot
+          // resurrect this wait as "done".
           deps.fleetRecords.interrupt(id, session.report);
           const taken = deps.fleetRecords.take(id);
           return {

src/subagent/authority.ts

@@ -6,11 +6,11 @@
  *
  *  - assertTierMayMountFleetVerb: a Tier 3 leaf may never mount a fleet verb
  *    (task, spawn_agent, wait_agents, list_agents, interrupt_agent, close_agent,
- *    resume_agent, followup_task, read_agent_trace, search_agents; reserved:
- *    send_input). Fleet *discovery* of the director catalog
- *    (search_agents) is Tier 1 only (CL-7051). list_agents is not catalog
- *    discovery — it lists this install's own spawn_agent workers, the same
- *    scoped mailbox wait_agents uses, so nested orchestrators may mount it.
+ *    resume_agent, followup_task, send_input, read_agent_trace, search_agents).
+ *    Fleet *discovery* of the director catalog (search_agents) is Tier 1 only
+ *    (CL-7051). list_agents is not catalog discovery — it lists this install's
+ *    own spawn_agent workers, the same scoped mailbox wait_agents uses, so
+ *    nested orchestrators may mount it.
  *  - assertCanTargetAgent: a Tier 2 nested orchestrator may act only on its
  *    own descendants, never a sibling or anything above it in the tree.
  *    Tier 1 (the primary orchestrator) may target anyone. Callers pass the
@@ -25,8 +25,7 @@ export type { SubagentTier } from "../agent/directors/types.js";
 
 /**
  * Every tool that grants control over other agents (spawn, list, steer,
- * observe). Tier 3 leaves may mount none of these — ever. Reserved names
- * `send_input` stays reserved so a later mount site inherits the gate.
+ * observe). Tier 3 leaves may mount none of these — ever.
  */
 export const FLEET_VERBS = new Set([
   "task",
@@ -108,15 +107,15 @@ function isDescendant(
 }
 
 /**
- * Live gate for `read_agent_trace` (and any future verb that addresses an
- * existing session). Callers that only spawn (`task`, `spawn_agent`) never
- * reach this check.
- *
  * Authority rule (root owns its tree; a child manages only its own
  * descendants): throws unless `actor` is Tier 1, or `targetId` is `actor.id`
  * itself, or a descendant of `actor.id` in `nodes`. A Tier 3 leaf holds no
  * fleet verbs at all and can never reach this check with a real call, so it
  * always fails closed here too.
+ *
+ * Production call sites: `read_agent_trace`, `send_input`, `interrupt_agent`,
+ * `close_agent`, `resume_agent`, and `followup_task` (nested mounts pass
+ * authority from run.ts; Tier-1 primary omits it and stays unrestricted).
  */
 export function assertCanTargetAgent(
   actor: { readonly id: string; readonly tier: SubagentTier },