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.

Author: TheGreatAxios

src/agent/fleet-verbs-mount.test.ts

@@ -17,6 +17,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

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

src/agent/tool-search.ts

@@ -49,6 +49,7 @@ export const CORE_TOOL_NAMES: readonly string[] = [
   "resume_agent",
   "interrupt_agent",
   "followup_task",
+  "send_input",
 ];
 
 const ORCHESTRATOR_ONLY_TOOL_NAMES: readonly string[] = [
@@ -60,6 +61,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

@@ -49,6 +49,7 @@ import {
   createResumeAgentTool,
   createInterruptAgentTool,
   createFollowupTaskTool,
+  createSendInputTool,
 } from "../subagent/lifecycle-tools.js";
 import { parseManageTasksArgs } from "./tasks.js";
 import { createListDirTool } from "../util/list-dir.js";
@@ -348,6 +349,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

@@ -7,7 +7,7 @@ import {
   MAX_FLEET_RECORDS,
   type AgentFleetDeps,
 } from "./agent-fleet.js";
-import { createInterruptAgentTool } from "./lifecycle-tools.js";
+import { createInterruptAgentTool, createSendInputTool } from "./lifecycle-tools.js";
 import { createSubAgentSessionStore } from "./session-store.js";
 import { createPermissionGate } from "../permission/gate.js";
 import { forcedStopReport } from "./stop-policy.js";
@@ -509,6 +509,7 @@ describe("interrupt_agent unblocks wait_agents", () => {
         close: async () => {},
         interrupt: () => {},
         followup: async () => "",
+        deliver: () => {},
       });
       return gate.promise;
     });
@@ -571,4 +572,74 @@ describe("interrupt_agent unblocks wait_agents", () => {
     expect(results[0]!.status).toBe("interrupted");
     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");
+  });
 });

src/subagent/agent-fleet.ts

@@ -151,6 +151,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()];
   }
@@ -516,10 +530,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);
         },
       };

src/subagent/authority.ts

@@ -6,11 +6,11 @@
  *
  *  - assertTierMayMountFleetVerb: a Tier 3 leaf may never mount a fleet verb
  *    (task, spawn_agent, wait_agents, interrupt_agent, close_agent,
- *    resume_agent, followup_task, read_agent_trace, search_agents; reserved:
- *    list_agents, send_input). Fleet *discovery* verbs (search_agents,
- *    list_agents) are further restricted to Tier 1 only (CL-7051) — nested
- *    orchestrators keep task/spawn allowlists but must not discover the
- *    full fleet.
+ *    resume_agent, followup_task, send_input, read_agent_trace,
+ *    search_agents; reserved: list_agents). Fleet *discovery* verbs
+ *    (search_agents, list_agents) are further restricted to Tier 1 only
+ *    (CL-7051) — nested orchestrators keep task/spawn allowlists but must
+ *    not discover the full fleet.
  *  - 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
@@ -26,8 +26,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
- * (`list_agents`, `send_input`) stay in the set so a later mount site
- * inherits the gate instead of needing a second allowlist.
+ * `list_agents` stays reserved so a later mount site inherits the gate.
  */
 export const FLEET_VERBS = new Set([
   "task",

src/subagent/lifecycle-tools.test.ts

@@ -5,6 +5,7 @@ import {
   createResumeAgentTool,
   createInterruptAgentTool,
   createFollowupTaskTool,
+  createSendInputTool,
 } from "./lifecycle-tools.js";
 import { createSubAgentSessionStore } from "./session-store.js";
 
@@ -13,7 +14,8 @@ async function callTool(
     | ReturnType<typeof createCloseAgentTool>
     | ReturnType<typeof createResumeAgentTool>
     | ReturnType<typeof createInterruptAgentTool>
-    | ReturnType<typeof createFollowupTaskTool>,
+    | ReturnType<typeof createFollowupTaskTool>
+    | ReturnType<typeof createSendInputTool>,
   args: Record<string, unknown>,
 ): Promise<Record<string, unknown>> {
   if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`);
@@ -258,3 +260,131 @@ describe("interrupt_agent / followup_task", () => {
     expect(followupErr.isError).toBe(true);
   });
 });
+
+describe("send_input", () => {
+  test("soft-delivers without flipping lifecycle or awaiting a reply", async () => {
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "worker",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(worker.id);
+    const delivered: string[] = [];
+    sessions.registerDeliver(worker.id, (message) => {
+      delivered.push(message);
+    });
+
+    const sendInput = createSendInputTool({ sessions });
+    const result = await callTool(sendInput, {
+      target: worker.id,
+      message: "stop and inspect line 4",
+    });
+
+    expect(result).toEqual({ agent_id: worker.id, status: "running" });
+    expect(delivered).toEqual(["stop and inspect line 4"]);
+    expect(sessions.get(worker.id)?.lifecycleStatus).toBe("running");
+  });
+
+  test("interrupt:true queues followup without awaiting and refuses when followup is missing", async () => {
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "worker",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(worker.id);
+    let interrupted = false;
+    let followupStarted = false;
+    sessions.registerInterrupt(worker.id, () => {
+      interrupted = true;
+    });
+    sessions.registerFollowup(worker.id, async (message) => {
+      followupStarted = true;
+      expect(message).toBe("patch only the test");
+      await new Promise((resolve) => setTimeout(resolve, 20));
+      return "queued turn finished";
+    });
+    sessions.registerDeliver(worker.id, () => {
+      throw new Error("interrupt:true should not soft-deliver");
+    });
+
+    const sendInput = createSendInputTool({ sessions });
+    const result = await callTool(sendInput, {
+      target: worker.id,
+      message: "patch only the test",
+      interrupt: true,
+    });
+    expect(result).toEqual({ agent_id: worker.id, status: "interrupted" });
+    expect(interrupted).toBe(true);
+    expect(followupStarted).toBe(true);
+    expect(sessions.get(worker.id)?.lifecycleStatus).toBe("interrupted");
+
+    const missing = sessions.start({
+      description: "no-followup",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(missing.id);
+    sessions.registerInterrupt(missing.id, () => {});
+    if (sendInput.kind !== "full") throw new Error("expected full tool");
+    const denied = await sendInput.handler(
+      {
+        id: "missing-followup",
+        name: "send_input",
+        arguments: { target: missing.id, message: "steer", interrupt: true },
+      },
+      new AbortController().signal,
+    );
+    expect(denied.isError).toBe(true);
+    expect(sessions.get(missing.id)?.lifecycleStatus).toBe("running");
+  });
+
+  test("enforces nested orchestrator descendant authority", async () => {
+    const sessions = createSubAgentSessionStore();
+    const nested = sessions.start({
+      id: "nested",
+      description: "nested",
+      agentId: "a",
+      brief: "b",
+    });
+    const child = sessions.start({
+      id: "child",
+      description: "child",
+      agentId: "a",
+      brief: "b",
+      parentSessionId: nested.id,
+    });
+    const sibling = sessions.start({
+      id: "sibling",
+      description: "sibling",
+      agentId: "a",
+      brief: "b",
+    });
+    for (const session of [nested, child, sibling]) {
+      sessions.markRunning(session.id);
+      sessions.registerDeliver(session.id, () => {});
+    }
+    const sendInput = createSendInputTool({
+      sessions,
+      authority: {
+        actorId: nested.id,
+        tier: "nested-orchestrator",
+        getNodes: () => sessions.list(),
+      },
+    });
+
+    const ok = await callTool(sendInput, { target: child.id, message: "continue" });
+    expect(ok.status).toBe("running");
+
+    if (sendInput.kind !== "full") throw new Error("expected full tool");
+    const denied = await sendInput.handler(
+      { id: "denied", name: "send_input", arguments: { target: sibling.id, message: "continue" } },
+      new AbortController().signal,
+    );
+    expect(denied.isError).toBe(true);
+  });
+});