diff --git a/src/agent/fleet-verbs-mount.test.ts b/src/agent/fleet-verbs-mount.test.ts index ac53419b7..267d7e59a 100644 --- a/src/agent/fleet-verbs-mount.test.ts +++ b/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 = { diff --git a/src/agent/tool-search.ts b/src/agent/tool-search.ts index 4d58ca801..e0a96ea20 100644 --- a/src/agent/tool-search.ts +++ b/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 diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 23a46b14b..7fbd25c3d 100644 --- a/src/agent/tools.ts +++ b/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 { + 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 42e8fc465..9fff87238 100644 --- a/src/subagent/authority.ts +++ b/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 diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index a853022d1..bd7183fed 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 { createSubAgentSessionStore } from "./session-store.js"; @@ -13,7 +14,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}`); @@ -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); + }); +}); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index 68601b556..284c44cc7 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -1,14 +1,10 @@ /** - * close_agent / resume_agent: the session-lifecycle half of - * reusable worker sessions. spawn_agent/wait_agents start and - * collect workers; these two verbs let an orchestrator tear one down on - * purpose (close_agent) or bring a retained one back for further input - * (resume_agent), instead of every session dying the instant its turn ends. - * - * interrupt_agent and followup_task (the verbs that actually push a new - * prompt into a resumed session) are a separate, later change — resume_agent - * here only flips a retained session back to an addressable state; it takes - * no prompt argument. + * close_agent / resume_agent / interrupt_agent / followup_task / send_input: + * the session-lifecycle half of reusable worker sessions. spawn_agent/ + * wait_agents start and collect workers; these verbs let an orchestrator tear + * one down on purpose (close_agent), bring a retained one back (resume_agent), + * stop a turn without teardown (interrupt_agent), push new work into a retained + * session (followup_task), or soft-/hard-steer a running worker (send_input). */ import { tool } from "@intx/agent"; @@ -17,7 +13,17 @@ import { type } from "arktype"; import type { ToolDefinition, ToolResult } from "@intx/types/runtime"; import { DEFAULT_CLOSE_DEADLINE_MS } from "./dispose.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,6 +94,20 @@ export interface LifecycleToolDeps { sessions: SubAgentSessionStore; } +/** + * Descendant-scoping for a Tier 2 nested orchestrator's `send_input`. + * Omit for Tier 1 (primary), which may target anyone. + */ +export interface SendInputAuthority { + actorId: string | undefined; + tier: SubagentTier; + getNodes: () => readonly FleetNode[]; +} + +export interface SendInputToolDeps extends LifecycleToolDeps { + authority?: SendInputAuthority; +} + export function createCloseAgentTool(deps: LifecycleToolDeps): AgentTool { return tool({ definition: closeAgentToolDefinition, @@ -142,7 +162,10 @@ export function createResumeAgentTool(deps: LifecycleToolDeps): AgentTool { `Error: cannot resume "${target}" (status: ${outcome.status}).${hint}`, ); } - return lifecycleResult(call.id, JSON.stringify({ agent_id: target, status: "running" })); + return lifecycleResult( + call.id, + JSON.stringify({ agent_id: target, status: "running" satisfies AgentLifecycleStatus }), + ); }, }); } @@ -249,3 +272,92 @@ 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): durable-deliver `message` into the " + + "worker's live session via agent.deliver and return immediately without awaiting a reply. " + + "With interrupt:true: stop the current turn (same as interrupt_agent) then queue `message` " + + "as the next-turn followup without awaiting that reply either — returns status " + + "'interrupted'. Fails on a session that is not currently running, or when the message is " + + "empty / oversize. Tier 2 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 " + + "(no await). 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 outcome = deps.sessions.sendInputOne(target, message, { + ...(parsed.interrupt === true ? { interrupt: true } : {}), + }); + if (!outcome.ok) { + return lifecycleResult( + call.id, + `Error: cannot send_input to "${target}" (status: ${outcome.status}).`, + ); + } + 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 9feb03660..bfeab55c2 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -112,6 +112,7 @@ import { createResumeAgentTool, createInterruptAgentTool, createFollowupTaskTool, + createSendInputTool, } from "./lifecycle-tools.js"; import { createSubAgentSessionStore } from "./session-store.js"; import type { RunSubAgentParams, RunSubAgentResult, SubAgentProvider } from "./types.js"; @@ -485,6 +486,7 @@ export async function runSubAgent(params: RunSubAgentParams): Promise fleetSessions.list(), + }, + }), ]; } @@ -852,7 +862,25 @@ 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..7968e27ac 100644 --- a/src/subagent/session-store.ts +++ b/src/subagent/session-store.ts @@ -202,6 +202,18 @@ export interface SubAgentSessionStore { ): Promise< { ok: true; reply: string } | { ok: false; status: AgentLifecycleStatus; hint?: string } >; + // CL-6944: registers the live agent.deliver handle so send_input can soft-steer + // a running worker (durable mid-run injection) without awaiting a reply. + registerDeliver(id: string, deliver: (message: string) => void): void; + // Soft: running only → deliver durable message, return immediately. + // interrupt:true → interruptOne then queue a next-turn followup without + // awaiting the reply. Fails closed when the target is not running / has no + // registered handle. + sendInputOne( + id: string, + message: string, + opts?: { interrupt?: boolean }, + ): { ok: true; status: AgentLifecycleStatus } | { ok: false; status: AgentLifecycleStatus }; subscribe(listener: () => void): () => void; clear(): void; } @@ -213,7 +225,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 +372,10 @@ export function createSubAgentSessionStore( // an interrupt can never accidentally resolve to the close codepath. const interruptHandles = new Map void>(); const followupHandles = new Map Promise>(); + // CL-6944: soft-steer deliver handles (agent.deliver), distinct from + // followupHandles (agent.send) so mid-run injection cannot be confused with + // a next-turn followup. + 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 @@ -463,6 +480,9 @@ export function createSubAgentSessionStore( void close(DEFAULT_CLOSE_DEADLINE_MS).catch(() => {}); } cancelHandles.delete(id); + interruptHandles.delete(id); + followupHandles.delete(id); + deliverHandles.delete(id); }; // An open retained session (spawn_agent's reusable-session contract: @@ -602,6 +622,7 @@ export function createSubAgentSessionStore( closeHandles.delete(id); interruptHandles.delete(id); followupHandles.delete(id); + deliverHandles.delete(id); forgetRevision(id); const session: SubAgentSession = { id, @@ -912,6 +933,7 @@ export function createSubAgentSessionStore( cancelHandles.delete(id); interruptHandles.delete(id); followupHandles.delete(id); + deliverHandles.delete(id); pruneCompleted(); return "shutdown"; }, @@ -926,6 +948,11 @@ export function createSubAgentSessionStore( followupHandles.set(id, followup); }, + registerDeliver(id: string, deliver: (message: string) => void): void { + if (!sessions.has(id)) return; + deliverHandles.set(id, deliver); + }, + interruptOne(id: string): { ok: true } | { ok: false; status: AgentLifecycleStatus } { const session = sessions.get(id); if (session === undefined) return { ok: false, status: "not_found" }; @@ -940,6 +967,54 @@ export function createSubAgentSessionStore( return { ok: true }; }, + sendInputOne( + id: string, + message: string, + opts?: { interrupt?: boolean }, + ): { 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"; + }); + // Queue the next-turn message on the same live agent without awaiting + // its reply — send_input returns immediately with "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) }); + }); + pruneRetained(); + }) + .catch(() => { + // Best-effort: a failed queued followup must not reject the + // already-returned send_input call. + }); + 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" }; + }, + async followupOne( id: string, message: string, @@ -1037,6 +1112,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 adfdb8c6c..ee641d938 100644 --- a/src/subagent/types.ts +++ b/src/subagent/types.ts @@ -172,6 +172,9 @@ export type RunSubAgentParams = { * active — this is the resume mechanism `resume_agent`/`followup_task` * build on, reusing `agent.send`'s own FIFO send-queue ordering rather * than a second continuation scheme. + * - `deliver`: durable mid-run injection via `agent.deliver` (not + * ephemeralTurns) so `send_input` can soft-steer a running worker + * without awaiting a reply. * * Always fired regardless of `persist`, so a caller can act on a * still-running session too, not only a retained one. The deadline @@ -183,6 +186,7 @@ export type RunSubAgentParams = { close: (deadlineMs?: number) => Promise; interrupt: () => void; followup: (message: string) => Promise; + deliver: (message: string) => void; }) => void; } & SubAgentSandboxDeps;