From fbd86659f41be04577f6edd63e765b8941f72a22 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 25 Aug 2026 08:52:59 -0700 Subject: [PATCH 1/3] 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. --- src/agent/fleet-verbs-mount.test.ts | 1 + src/agent/tool-search.test.ts | 2 + src/agent/tool-search.ts | 2 + src/agent/tools.ts | 2 + src/subagent/agent-fleet.test.ts | 81 +++++++++++++++- src/subagent/agent-fleet.ts | 17 +++- src/subagent/authority.ts | 13 ++- src/subagent/lifecycle-tools.test.ts | 132 ++++++++++++++++++++++++++- src/subagent/lifecycle-tools.ts | 122 ++++++++++++++++++++++++- src/subagent/run.ts | 28 +++++- src/subagent/session-store.ts | 86 ++++++++++++++++- src/subagent/types.ts | 1 + 12 files changed, 471 insertions(+), 16 deletions(-) diff --git a/src/agent/fleet-verbs-mount.test.ts b/src/agent/fleet-verbs-mount.test.ts index db0333c6f..ac027483a 100644 --- a/src/agent/fleet-verbs-mount.test.ts +++ b/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", () => { diff --git a/src/agent/tool-search.test.ts b/src/agent/tool-search.test.ts index 26c0a0b46..d03baa278 100644 --- a/src/agent/tool-search.test.ts +++ b/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); } diff --git a/src/agent/tool-search.ts b/src/agent/tool-search.ts index 0c5f873f2..acb1856fb 100644 --- a/src/agent/tool-search.ts +++ b/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 diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 7c5f5e605..1055a3933 100644 --- a/src/agent/tools.ts +++ b/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 { 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,6 +637,76 @@ 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(); + 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(); + const followupGate = deferred(); + 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(); const deps = makeDeps(async (params) => { @@ -638,6 +714,7 @@ describe("interrupt_agent unblocks wait_agents", () => { 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; }); @@ -732,6 +810,7 @@ describe("close_agent unblocks wait_agents", () => { close: async () => {}, interrupt: () => {}, followup: async () => "", + deliver: () => {}, }); return gate.promise; }); diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index a3436ed87..1d3ab5a0f 100644 --- a/src/subagent/agent-fleet.ts +++ b/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); }, }; diff --git a/src/subagent/authority.ts b/src/subagent/authority.ts index 032f11dfd..1144f9ec3 100644 --- a/src/subagent/authority.ts +++ b/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", diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index 7061db7ac..2294607df 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -5,6 +5,7 @@ import { createResumeAgentTool, createInterruptAgentTool, createFollowupTaskTool, + createSendInputTool, } from "./lifecycle-tools.js"; import { createFleetRecords } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; @@ -14,7 +15,8 @@ async function callTool( | ReturnType | ReturnType | ReturnType - | ReturnType, + | ReturnType + | ReturnType, args: Record, ): Promise> { if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`); @@ -268,3 +270,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); + }); +}); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index 59cecba8e..c39abbe89 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -18,7 +18,17 @@ import type { ToolDefinition, ToolResult } from "@intx/types/runtime"; import { DEFAULT_CLOSE_DEADLINE_MS } from "./dispose.js"; import type { FleetRecordsHandle } from "./agent-fleet.js"; -import type { AgentLifecycleStatus, SubAgentSessionStore } from "./session-store.js"; +import { + DEFAULT_MAX_ENTRY_CHARS, + type AgentLifecycleStatus, + type SubAgentSessionStore, +} from "./session-store.js"; +import { + assertCanTargetAgent, + FleetAuthorityError, + type FleetNode, + type SubagentTier, +} from "./authority.js"; function lifecycleResult(callId: string, content: string): ToolResult { const isError = content.startsWith("Error:"); @@ -88,7 +98,7 @@ function descendantsClosingOrder( export interface LifecycleToolDeps { sessions: SubAgentSessionStore; - /** Optional for resume/followup; close and interrupt require it (see CloseAgentToolDeps / InterruptAgentToolDeps). */ + /** Optional for resume/followup/send_input; close and interrupt require it (see CloseAgentToolDeps / InterruptAgentToolDeps). */ fleetRecords?: FleetRecordsHandle; } @@ -102,6 +112,16 @@ export type InterruptAgentToolDeps = LifecycleToolDeps & { fleetRecords: FleetRecordsHandle; }; +export interface SendInputAuthority { + actorId: string | undefined; + tier: SubagentTier; + getNodes: () => readonly FleetNode[]; +} + +export interface SendInputToolDeps extends LifecycleToolDeps { + authority?: SendInputAuthority; +} + export function createCloseAgentTool(deps: CloseAgentToolDeps): AgentTool { return tool({ definition: closeAgentToolDefinition, @@ -272,3 +292,101 @@ export function createFollowupTaskTool(deps: LifecycleToolDeps): AgentTool { }, }); } + +const SendInputArgs = type({ + target: "string", + message: "string", + "interrupt?": "boolean", +}); + +export const sendInputToolDefinition: ToolDefinition = { + name: "send_input", + description: + "Steer a running worker mid-turn. Soft (default): deliver `message` into the live session " + + "and return immediately without awaiting a reply and without completing wait_agents. " + + "With interrupt:true: stop the current turn (same wait-mailbox flip as interrupt_agent) " + + "then queue `message` as the next-turn followup without awaiting that reply. Fails on a " + + "session that is not currently running, or when the message is empty / oversize. Nested " + + "orchestrators may only target their own descendants.", + inputSchema: { + type: "object", + properties: { + target: { type: "string", description: "agent_id of the running session to steer." }, + message: { + type: "string", + description: `Instruction to inject (non-empty, max ${DEFAULT_MAX_ENTRY_CHARS} characters).`, + }, + interrupt: { + type: "boolean", + description: + "When true, interrupt the current turn then queue message as the next-turn followup. " + + "When false/omitted, soft-deliver into the running turn.", + }, + }, + required: ["target", "message"], + }, +}; + +export function createSendInputTool(deps: SendInputToolDeps): AgentTool { + return tool({ + definition: sendInputToolDefinition, + handler: async (call, _signal): Promise => { + const parsed = SendInputArgs(call.arguments); + if (parsed instanceof type.errors) { + return lifecycleResult(call.id, `Error: send_input arguments invalid: ${parsed.summary}`); + } + const target = parsed.target.trim(); + const message = parsed.message.trim(); + if (message.length === 0) { + return lifecycleResult(call.id, "Error: send_input requires a non-empty message."); + } + if (message.length > DEFAULT_MAX_ENTRY_CHARS) { + return lifecycleResult( + call.id, + `Error: send_input message exceeds ${DEFAULT_MAX_ENTRY_CHARS} characters ` + + `(got ${message.length}).`, + ); + } + if (deps.authority !== undefined) { + if (deps.authority.actorId === undefined) { + return lifecycleResult( + call.id, + "Error: send_input is unavailable for this worker (no resolvable session " + + "id to scope descendant access).", + ); + } + try { + assertCanTargetAgent( + { id: deps.authority.actorId, tier: deps.authority.tier }, + target, + deps.authority.getNodes(), + ); + } catch (cause) { + if (cause instanceof FleetAuthorityError) { + return lifecycleResult(call.id, `Error: ${cause.message}`); + } + throw cause; + } + } + const interrupt = parsed.interrupt === true; + const outcome = deps.sessions.sendInputOne(target, message, { + ...(interrupt ? { interrupt: true } : {}), + ...(interrupt && deps.fleetRecords !== undefined + ? { + onFollowupReply: (reply: string) => { + deps.fleetRecords?.completeAfterInterrupt(target, reply); + }, + } + : {}), + }); + if (!outcome.ok) { + return lifecycleResult( + call.id, + `Error: cannot send_input to "${target}" (status: ${outcome.status}).`, + ); + } + if (interrupt) deps.fleetRecords?.interrupt(target); + return lifecycleResult(call.id, JSON.stringify({ agent_id: target, status: outcome.status })); + }, + }); +} diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 76292dd4a..b0547fe5b 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -118,6 +118,7 @@ import { createResumeAgentTool, createInterruptAgentTool, createFollowupTaskTool, + createSendInputTool, } from "./lifecycle-tools.js"; import { createSubAgentSessionStore } from "./session-store.js"; import type { RunSubAgentParams, RunSubAgentResult, SubAgentProvider } from "./types.js"; @@ -510,6 +511,7 @@ export async function runSubAgent(params: RunSubAgentParams): Promise fleetSessions.list(), + }, + }), ]; } @@ -903,7 +914,22 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { + agent!.deliver({ + ref: { uid: 1, mailbox: "INBOX" }, + headers: { + from: "parent@local", + to: ["agent@local"], + date: new Date().toISOString(), + messageId: ``, + interchangeType: "conversation.message", + }, + flags: [], + content: message, + signatureStatus: "missing", + }); + }; + params.onAgentReady({ close: boundedClose, interrupt, followup, deliver }); } const fullPrompt = buildDispatchBrief({ diff --git a/src/subagent/session-store.ts b/src/subagent/session-store.ts index bbf8ab8f4..0ad9fe806 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -5,10 +5,14 @@ // this store is the dedicated child record the enter-session UI reads. import type { ReactorEmittedEvent } from "@intx/inference"; +import { getLogger } from "@intx/log"; +import { LOG_NAMESPACE_ROOT } from "../branding.js"; import { DEFAULT_CLOSE_DEADLINE_MS } from "./dispose.js"; import type { ForcedStopReason } from "./stop-policy.js"; import { toolCallPreview } from "./tool-preview.js"; +const log = getLogger([LOG_NAMESPACE_ROOT, "subagent", "session-store"]); + export type SubAgentSessionStatus = "running" | "done" | "failed" | "cancelled"; /** @@ -202,6 +206,12 @@ export interface SubAgentSessionStore { ): Promise< { ok: true; reply: string } | { ok: false; status: AgentLifecycleStatus; hint?: string } >; + registerDeliver(id: string, deliver: (message: string) => void): void; + sendInputOne( + id: string, + message: string, + opts?: { interrupt?: boolean; onFollowupReply?: (reply: string) => void }, + ): { ok: true; status: AgentLifecycleStatus } | { ok: false; status: AgentLifecycleStatus }; subscribe(listener: () => void): () => void; clear(): void; } @@ -213,7 +223,8 @@ const DEFAULT_MAX_COMPLETED = 20; // sidebar list — see maxRetained doc above. const DEFAULT_MAX_RETAINED = 50; const DEFAULT_MAX_ENTRIES = 400; -const DEFAULT_MAX_ENTRY_CHARS = 24_000; +/** Cap on characters per transcript entry / send_input message body. */ +export const DEFAULT_MAX_ENTRY_CHARS = 24_000; const EVICTED_RETENTION_HINT = "Session evicted to bound retained-session memory; recover full detail via read_agent_trace(agent_id)."; @@ -359,6 +370,7 @@ export function createSubAgentSessionStore( // an interrupt can never accidentally resolve to the close codepath. const interruptHandles = new Map void>(); const followupHandles = new Map Promise>(); + const deliverHandles = new Map void>(); const listeners = new Set<() => void>(); // CL-7007: tombstones for sessions dropped by pruneRetained, keyed by id, // insertion-ordered (Map preserves it) so the oldest can be dropped first @@ -460,9 +472,16 @@ export function createSubAgentSessionStore( const close = closeHandles.get(id); if (close !== undefined) { closeHandles.delete(id); - void close(DEFAULT_CLOSE_DEADLINE_MS).catch(() => {}); + void close(DEFAULT_CLOSE_DEADLINE_MS).catch((err: unknown) => { + log.warn("session close during handle release failed: {error}", { + error: err instanceof Error ? err.message : String(err), + }); + }); } cancelHandles.delete(id); + interruptHandles.delete(id); + followupHandles.delete(id); + deliverHandles.delete(id); }; // An open retained session (spawn_agent's reusable-session contract: @@ -602,6 +621,7 @@ export function createSubAgentSessionStore( closeHandles.delete(id); interruptHandles.delete(id); followupHandles.delete(id); + deliverHandles.delete(id); forgetRevision(id); const session: SubAgentSession = { id, @@ -897,7 +917,11 @@ export function createSubAgentSessionStore( // close that does not honor its own deadline argument — a wedged // descendant must not hang the whole close_agent call. await Promise.race([ - close(deadlineMs).catch(() => {}), + close(deadlineMs).catch((err: unknown) => { + log.warn("session close raced deadline: {error}", { + error: err instanceof Error ? err.message : String(err), + }); + }), new Promise((resolve) => setTimeout(resolve, deadlineMs)), ]); mutate(id, (s) => { @@ -912,6 +936,7 @@ export function createSubAgentSessionStore( cancelHandles.delete(id); interruptHandles.delete(id); followupHandles.delete(id); + deliverHandles.delete(id); pruneCompleted(); return "shutdown"; }, @@ -926,6 +951,60 @@ export function createSubAgentSessionStore( followupHandles.set(id, followup); }, + registerDeliver(id: string, deliver: (message: string) => void): void { + if (!sessions.has(id)) return; + deliverHandles.set(id, deliver); + }, + + sendInputOne( + id: string, + message: string, + opts?: { interrupt?: boolean; onFollowupReply?: (reply: string) => void }, + ): { ok: true; status: AgentLifecycleStatus } | { ok: false; status: AgentLifecycleStatus } { + const session = sessions.get(id); + if (session === undefined) return { ok: false, status: "not_found" }; + if (session.status !== "running") return { ok: false, status: session.lifecycleStatus }; + + if (opts?.interrupt === true) { + const interrupt = interruptHandles.get(id); + const followup = followupHandles.get(id); + if (interrupt === undefined || followup === undefined) { + return { ok: false, status: session.lifecycleStatus }; + } + interrupt(); + mutate(id, (s) => { + s.lifecycleStatus = "interrupted"; + }); + void followup(message) + .then((reply) => { + const still = sessions.get(id); + if (still === undefined || still.lifecycleStatus === "shutdown") return; + mutate(id, (s) => { + s.status = "done"; + s.lifecycleStatus = "completed"; + s.finishedAt = now(); + s.report = reply; + pushEntry(s, { kind: "report", content: capText(reply, maxEntryChars) }); + }); + opts.onFollowupReply?.(reply); + pruneRetained(); + }) + .catch((err: unknown) => { + log.error("send_input followup failed for {id}: {error}", { + id, + error: err instanceof Error ? err.message : String(err), + }); + }); + pruneRetained(); + return { ok: true, status: "interrupted" }; + } + + const deliver = deliverHandles.get(id); + if (deliver === undefined) return { ok: false, status: session.lifecycleStatus }; + deliver(message); + return { ok: true, status: "running" }; + }, + interruptOne(id: string): { ok: true } | { ok: false; status: AgentLifecycleStatus } { const session = sessions.get(id); if (session === undefined) return { ok: false, status: "not_found" }; @@ -1037,6 +1116,7 @@ export function createSubAgentSessionStore( closeHandles.clear(); interruptHandles.clear(); followupHandles.clear(); + deliverHandles.clear(); sessions.clear(); revisions.clear(); snapshotCache.clear(); diff --git a/src/subagent/types.ts b/src/subagent/types.ts index 129533079..b0abc88c4 100644 --- a/src/subagent/types.ts +++ b/src/subagent/types.ts @@ -183,6 +183,7 @@ export type RunSubAgentParams = { close: (deadlineMs?: number) => Promise; interrupt: () => void; followup: (message: string) => Promise; + deliver: (message: string) => void; }) => void; } & SubAgentSandboxDeps; From 99c8780fd4fd4851fd2e7391fb083f52ade68e7b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 25 Aug 2026 08:31:49 -0700 Subject: [PATCH 2/3] 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. --- docs/ARCHITECTURE.md | 4 +- src/subagent/agent-fleet.test.ts | 29 +++++ src/subagent/agent-fleet.ts | 5 +- src/subagent/authority.ts | 8 +- src/subagent/lifecycle-tools.test.ts | 167 +++++++++++++++++++++++++++ src/subagent/lifecycle-tools.ts | 80 ++++++++----- src/subagent/run.ts | 23 ++-- 7 files changed, 269 insertions(+), 47 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e14f597af..3978bcd22 100644 --- a/docs/ARCHITECTURE.md +++ b/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/`) diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index d25feeb40..5fb0fe411 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -800,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", () => { diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 1d3ab5a0f..dd417b412 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -721,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 { diff --git a/src/subagent/authority.ts b/src/subagent/authority.ts index 1144f9ec3..33fbfbd4f 100644 --- a/src/subagent/authority.ts +++ b/src/subagent/authority.ts @@ -107,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 }, diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index 2294607df..10ee695be 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -397,4 +397,171 @@ describe("send_input", () => { ); expect(denied.isError).toBe(true); }); + + test("fails closed when nested authority has no actorId", async () => { + const sessions = createSubAgentSessionStore(); + const worker = sessions.start({ + id: "worker", + description: "worker", + agentId: "a", + brief: "b", + }); + sessions.markRunning(worker.id); + sessions.registerDeliver(worker.id, () => {}); + const sendInput = createSendInputTool({ + sessions, + authority: { + actorId: undefined, + tier: "nested-orchestrator", + getNodes: () => sessions.list(), + }, + }); + if (sendInput.kind !== "full") throw new Error("expected full tool"); + const denied = await sendInput.handler( + { id: "no-actor", name: "send_input", arguments: { target: worker.id, message: "x" } }, + new AbortController().signal, + ); + expect(denied.isError).toBe(true); + expect(String(denied.content)).toContain("no resolvable session"); + }); +}); + +describe("nested lifecycle authority", () => { + function nestAuthority(sessions: ReturnType, actorId: string) { + return { + actorId, + tier: "nested-orchestrator" as const, + getNodes: () => sessions.list(), + }; + } + + test("interrupt_agent denies a sibling and allows a descendant", async () => { + const sessions = createSubAgentSessionStore(); + const nested = sessions.start({ id: "nested", description: "n", agentId: "a", brief: "b" }); + const child = sessions.start({ + id: "child", + description: "c", + agentId: "a", + brief: "b", + parentSessionId: nested.id, + }); + const sibling = sessions.start({ id: "sibling", description: "s", agentId: "a", brief: "b" }); + for (const s of [child, sibling]) { + sessions.markRunning(s.id); + sessions.registerInterrupt(s.id, () => {}); + } + const interrupt = createInterruptAgentTool({ + sessions, + fleetRecords: createFleetRecords(), + authority: nestAuthority(sessions, nested.id), + }); + expect((await callTool(interrupt, { target: child.id })).status).toBe("interrupted"); + if (interrupt.kind !== "full") throw new Error("expected full tool"); + const denied = await interrupt.handler( + { id: "d", name: "interrupt_agent", arguments: { target: sibling.id } }, + new AbortController().signal, + ); + expect(denied.isError).toBe(true); + }); + + test("close_agent denies a sibling and allows a descendant", async () => { + const sessions = createSubAgentSessionStore(); + const nested = sessions.start({ id: "nested", description: "n", agentId: "a", brief: "b" }); + const child = sessions.start({ + id: "child", + description: "c", + agentId: "a", + brief: "b", + parentSessionId: nested.id, + }); + const sibling = sessions.start({ id: "sibling", description: "s", agentId: "a", brief: "b" }); + for (const s of [child, sibling]) sessions.registerClose(s.id, async () => {}); + const close = createCloseAgentTool({ + sessions, + fleetRecords: createFleetRecords(), + authority: nestAuthority(sessions, nested.id), + }); + expect((await callTool(close, { target: child.id })).status).toBe("shutdown"); + if (close.kind !== "full") throw new Error("expected full tool"); + const denied = await close.handler( + { id: "d", name: "close_agent", arguments: { target: sibling.id } }, + new AbortController().signal, + ); + expect(denied.isError).toBe(true); + expect(sessions.get(sibling.id)?.lifecycleStatus).not.toBe("shutdown"); + }); + + test("followup_task denies a sibling and allows a descendant", async () => { + const sessions = createSubAgentSessionStore(); + const nested = sessions.start({ id: "nested", description: "n", agentId: "a", brief: "b" }); + const child = sessions.start({ + id: "child", + description: "c", + agentId: "a", + brief: "b", + parentSessionId: nested.id, + retained: true, + }); + const sibling = sessions.start({ + id: "sibling", + description: "s", + agentId: "a", + brief: "b", + retained: true, + }); + for (const s of [child, sibling]) { + sessions.complete(s.id, "done"); + sessions.registerFollowup(s.id, async () => "reply"); + } + const followup = createFollowupTaskTool({ + sessions, + authority: nestAuthority(sessions, nested.id), + }); + expect((await callTool(followup, { target: child.id, message: "more" })).status).toBe( + "completed", + ); + if (followup.kind !== "full") throw new Error("expected full tool"); + const denied = await followup.handler( + { + id: "d", + name: "followup_task", + arguments: { target: sibling.id, message: "more" }, + }, + new AbortController().signal, + ); + expect(denied.isError).toBe(true); + }); + + test("resume_agent denies a sibling and allows a descendant", async () => { + const sessions = createSubAgentSessionStore(); + const nested = sessions.start({ id: "nested", description: "n", agentId: "a", brief: "b" }); + const child = sessions.start({ + id: "child", + description: "c", + agentId: "a", + brief: "b", + parentSessionId: nested.id, + retained: true, + }); + const sibling = sessions.start({ + id: "sibling", + description: "s", + agentId: "a", + brief: "b", + retained: true, + }); + sessions.complete(child.id, "done"); + sessions.complete(sibling.id, "done"); + const resume = createResumeAgentTool({ + sessions, + authority: nestAuthority(sessions, nested.id), + }); + expect((await callTool(resume, { target: child.id })).status).toBe("running"); + if (resume.kind !== "full") throw new Error("expected full tool"); + const denied = await resume.handler( + { id: "d", name: "resume_agent", arguments: { target: sibling.id } }, + new AbortController().signal, + ); + expect(denied.isError).toBe(true); + }); }); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index c39abbe89..a0adbdf8c 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -96,10 +96,22 @@ function descendantsClosingOrder( return order; } +/** + * Nested-orchestrator subtree gate for addressing verbs. When `authority` is + * omitted (Tier-1 primary mount), targeting is unrestricted. When present, + * a missing `actorId` fails closed — same rule as read_agent_trace. + */ +export interface LifecycleAuthority { + actorId: string | undefined; + tier: SubagentTier; + getNodes: () => readonly FleetNode[]; +} + export interface LifecycleToolDeps { sessions: SubAgentSessionStore; /** Optional for resume/followup/send_input; close and interrupt require it (see CloseAgentToolDeps / InterruptAgentToolDeps). */ fleetRecords?: FleetRecordsHandle; + authority?: LifecycleAuthority; } /** close_agent always terminalizes the wait mailbox — no silent skip. */ @@ -112,14 +124,33 @@ export type InterruptAgentToolDeps = LifecycleToolDeps & { fleetRecords: FleetRecordsHandle; }; -export interface SendInputAuthority { - actorId: string | undefined; - tier: SubagentTier; - getNodes: () => readonly FleetNode[]; -} - -export interface SendInputToolDeps extends LifecycleToolDeps { - authority?: SendInputAuthority; +function gateTarget( + deps: LifecycleToolDeps, + toolName: string, + target: string, + callId: string, +): ToolResult | undefined { + if (deps.authority === undefined) return undefined; + if (deps.authority.actorId === undefined) { + return lifecycleResult( + callId, + `Error: ${toolName} is unavailable for this worker (no resolvable session ` + + "id to scope descendant access).", + ); + } + try { + assertCanTargetAgent( + { id: deps.authority.actorId, tier: deps.authority.tier }, + target, + deps.authority.getNodes(), + ); + } catch (cause) { + if (cause instanceof FleetAuthorityError) { + return lifecycleResult(callId, `Error: ${cause.message}`); + } + throw cause; + } + return undefined; } export function createCloseAgentTool(deps: CloseAgentToolDeps): AgentTool { @@ -131,6 +162,8 @@ export function createCloseAgentTool(deps: CloseAgentToolDeps): AgentTool { return lifecycleResult(call.id, `Error: close_agent arguments invalid: ${parsed.summary}`); } const target = parsed.target.trim(); + const denied = gateTarget(deps, "close_agent", target, call.id); + if (denied !== undefined) return denied; if (deps.sessions.get(target) === undefined) { return lifecycleResult( call.id, @@ -173,6 +206,8 @@ export function createResumeAgentTool(deps: LifecycleToolDeps): AgentTool { return lifecycleResult(call.id, `Error: resume_agent arguments invalid: ${parsed.summary}`); } const target = parsed.target.trim(); + const denied = gateTarget(deps, "resume_agent", target, call.id); + if (denied !== undefined) return denied; const outcome = deps.sessions.resumeOne(target); if (!outcome.ok) { const hint = outcome.hint !== undefined ? ` ${outcome.hint}` : ""; @@ -221,6 +256,8 @@ export function createInterruptAgentTool(deps: InterruptAgentToolDeps): AgentToo ); } const target = parsed.target.trim(); + const denied = gateTarget(deps, "interrupt_agent", target, call.id); + if (denied !== undefined) return denied; const outcome = deps.sessions.interruptOne(target); if (!outcome.ok) { return lifecycleResult( @@ -273,6 +310,8 @@ export function createFollowupTaskTool(deps: LifecycleToolDeps): AgentTool { ); } const target = parsed.target.trim(); + const denied = gateTarget(deps, "followup_task", target, call.id); + if (denied !== undefined) return denied; const message = parsed.message.trim(); if (message.length === 0) { return lifecycleResult(call.id, "Error: followup_task requires a non-empty message."); @@ -327,7 +366,7 @@ export const sendInputToolDefinition: ToolDefinition = { }, }; -export function createSendInputTool(deps: SendInputToolDeps): AgentTool { +export function createSendInputTool(deps: LifecycleToolDeps): AgentTool { return tool({ definition: sendInputToolDefinition, handler: async (call, _signal): Promise => { @@ -336,6 +375,8 @@ export function createSendInputTool(deps: SendInputToolDeps): AgentTool { return lifecycleResult(call.id, `Error: send_input arguments invalid: ${parsed.summary}`); } const target = parsed.target.trim(); + const denied = gateTarget(deps, "send_input", target, call.id); + if (denied !== undefined) return denied; const message = parsed.message.trim(); if (message.length === 0) { return lifecycleResult(call.id, "Error: send_input requires a non-empty message."); @@ -347,27 +388,6 @@ export function createSendInputTool(deps: SendInputToolDeps): AgentTool { `(got ${message.length}).`, ); } - if (deps.authority !== undefined) { - if (deps.authority.actorId === undefined) { - return lifecycleResult( - call.id, - "Error: send_input is unavailable for this worker (no resolvable session " + - "id to scope descendant access).", - ); - } - try { - assertCanTargetAgent( - { id: deps.authority.actorId, tier: deps.authority.tier }, - target, - deps.authority.getNodes(), - ); - } catch (cause) { - if (cause instanceof FleetAuthorityError) { - return lifecycleResult(call.id, `Error: ${cause.message}`); - } - throw cause; - } - } const interrupt = parsed.interrupt === true; const outcome = deps.sessions.sendInputOne(target, message, { ...(interrupt ? { interrupt: true } : {}), diff --git a/src/subagent/run.ts b/src/subagent/run.ts index b0547fe5b..808075cc9 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -579,6 +579,11 @@ export async function runSubAgent(params: RunSubAgentParams): Promise fleetSessions.list(), + }; const fleetDeps = { permissionGate: nd.permissionGate, ...(nd.inheritMcpTools !== undefined ? { inheritMcpTools: nd.inheritMcpTools } : {}), @@ -604,18 +609,18 @@ export async function runSubAgent(params: RunSubAgentParams): Promise fleetSessions.list(), - }, + authority: lifecycleAuthority, }), ]; } From 8ab43d8a65b4e72c54d7b572fbb926b1edf43c55 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 25 Aug 2026 09:06:56 -0700 Subject: [PATCH 3/3] Format run.ts after send_input authority restack --- src/subagent/run.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 808075cc9..3a82c05e6 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -609,7 +609,11 @@ export async function runSubAgent(params: RunSubAgentParams): Promise