Add parent steering for running workers

Author: TheGreatAxios

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

@@ -1,7 +1,7 @@
 /**
- * Primary createAgentToolset mounts the six fleet verbs beside task /
- * search_agents / read_agent_trace when subAgent (with the shared TUI
- * sessions store) is wired. Leaves / no-subAgent toolsets stay without them.
+ * Primary createAgentToolset mounts the fleet verbs beside task / search_agents /
+ * read_agent_trace when subAgent (with the shared TUI sessions store) is wired.
+ * Leaves / no-subAgent toolsets stay without them.
  */
 import { mkdtempSync } from "node:fs";
 import { tmpdir } from "node:os";
@@ -17,10 +17,11 @@ const FLEET_VERBS = [
   "resume_agent",
   "interrupt_agent",
   "followup_task",
+  "send_input",
 ] as const;
 
 describe("primary fleet verb mount", () => {
-  test("createAgentToolset registers the six fleet verbs when subAgent + sessions are set", async () => {
+  test("createAgentToolset registers the fleet verbs when subAgent + sessions are set", async () => {
     const cwd = mkdtempSync(join(tmpdir(), "corbits-fleet-mount-"));
     const { createAgentToolset } = await import("./tools.js");
     const permissionGate = {

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,8 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise<AgentT
         createResumeAgentTool({ sessions: fleetSessions }),
         createInterruptAgentTool({ sessions: fleetSessions }),
         createFollowupTaskTool({ sessions: fleetSessions }),
+        // Tier 1: primary may target any worker — omit authority (unrestricted).
+        createSendInputTool({ sessions: fleetSessions }),
       );
     }
   }

src/subagent/agent-fleet.ts

@@ -449,10 +449,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

@@ -11,10 +11,11 @@
  *    this same gate).
  *  - 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
- *    live fleet as a flat list of {id, parentSessionId} nodes — the same
- *    shape SubAgentSessionStore already tracks — so no parallel tree
- *    structure is needed.
+ *    Tier 1 (the primary orchestrator) may target anyone. Addressing verbs
+ *    such as read_agent_trace and send_input call this at their handler
+ *    boundary. Callers pass the live fleet as a flat list of {id,
+ *    parentSessionId} nodes — the same shape SubAgentSessionStore already
+ *    tracks — so no parallel tree structure is needed.
  */
 
 import type { SubagentTier } from "../agent/directors/types.js";
@@ -89,15 +90,6 @@ function isDescendant(
 }
 
 /**
- * SEAM, NOT YET A LIVE GATE: this function has no production call site today.
- * No verb in this codebase currently lets one live agent target another
- * (`task` only spawns; it never addresses an existing session), so the
- * subtree rule below is exercised only by authority.test.ts — it is not
- * enforced at runtime yet. It exists now so future verbs that make one
- * agent addressable by another can call it from day one instead of
- * inventing their own check. Until one of those wires a call site here, do
- * not describe this rule as enforced; only assertTierMayMountFleetVerb is.
- *
  * 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

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,172 @@ describe("interrupt_agent / followup_task", () => {
     expect(followupErr.isError).toBe(true);
   });
 });
+
+describe("send_input", () => {
+  test("soft-delivers a durable message to a running worker without 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("interrupts and queues a next-turn message without awaiting the reply", 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("drop the broad refactor and patch only the test");
+      await new Promise((resolve) => setTimeout(resolve, 25));
+      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: "drop the broad refactor and 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");
+  });
+
+  test("fails closed when interrupt:true cannot queue the followup", async () => {
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "worker",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(worker.id);
+    let interrupted = false;
+    sessions.registerInterrupt(worker.id, () => {
+      interrupted = true;
+    });
+
+    const sendInput = createSendInputTool({ sessions });
+    if (sendInput.kind !== "full") throw new Error("expected full tool");
+    const result = await sendInput.handler(
+      {
+        id: "missing-followup",
+        name: "send_input",
+        arguments: { target: worker.id, message: "steer after interrupt", interrupt: true },
+      },
+      new AbortController().signal,
+    );
+
+    expect(result.isError).toBe(true);
+    expect(interrupted).toBe(false);
+    expect(sessions.get(worker.id)?.lifecycleStatus).toBe("running");
+  });
+
+  test("rejects empty and oversize messages", async () => {
+    const sessions = createSubAgentSessionStore();
+    const worker = sessions.start({
+      description: "worker",
+      agentId: "a",
+      brief: "b",
+      retained: true,
+    });
+    sessions.markRunning(worker.id);
+    sessions.registerDeliver(worker.id, () => {});
+    const sendInput = createSendInputTool({ sessions });
+
+    if (sendInput.kind !== "full") throw new Error("expected full tool");
+    const empty = await sendInput.handler(
+      { id: "empty", name: "send_input", arguments: { target: worker.id, message: "   " } },
+      new AbortController().signal,
+    );
+    expect(empty.isError).toBe(true);
+
+    const oversize = await sendInput.handler(
+      {
+        id: "big",
+        name: "send_input",
+        arguments: { target: worker.id, message: "x".repeat(24_001) },
+      },
+      new AbortController().signal,
+    );
+    expect(oversize.isError).toBe(true);
+  });
+
+  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);
+  });
+});