diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 20e93e6a0..634619a35 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -195,11 +195,11 @@ Invocation: workflows are **not** top-level slash commands. Recipe definitions l Three distinct concepts (do not conflate them): -| Concept | What it is | Surface | -| ------------- | -------------------------------------------------------- | ------------------------------------------------------------------- | -| **Agent** | A runtime entity with its own loop, tools, and context | Primary session or a spawned child | -| **Task** | A checklist item owned by _one_ agent via `manage_tasks` | Local work plan — not a spawn | -| **Sub-agent** | A short-lived child agent for one self-contained job | Spawned with the **`task`** tool (wire name kept for compatibility) | +| Concept | What it is | Surface | +| ------------- | -------------------------------------------------------- | --------------------------------------------------------- | +| **Agent** | A runtime entity with its own loop, tools, and context | Primary session or a spawned child | +| **Task** | A checklist item owned by _one_ agent via `manage_tasks` | Local work plan — not a spawn | +| **Sub-agent** | A short-lived child agent for one self-contained job | Spawned with **`spawn_agent`** (or deprecated **`task`**) | The **`task`** tool **spawns a sub-agent** on a separate inference source (tier/profile resolved from settings). The dispatch brief separates durable `context`, actionable `prompt`, and optional `goals` (checklist seeds for the _child's_ own `manage_tasks` list). The child returns a structured report (`Summary` / `Findings` / `Blockers` / `Paths`) plus a tools-used footer. Parent and child never share a `manage_tasks` list. @@ -219,9 +219,9 @@ 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 `task` / `search_agents`, 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 (CL-6942/CL-6944 can add one when a real caller needs it). `FLEET_VERBS` in `authority.ts` also names the not-yet-implemented verbs (`spawn_agent`, `wait_agents`, `list_agents`, `send_input`, `interrupt_agent`, `close_agent`, `resume_agent`, `read_agent_trace`, `followup_task`) so their future mount sites inherit the same gate. -- **Subtree authority — a seam, not yet wired.** `assertCanTargetAgent(actor, targetId, nodes)` (`src/subagent/authority.ts`) implements the "root owns its tree; a child manages only its own descendants" rule (Tier 1 may target anyone, Tier 2 may target only its own descendants over the same `{id, parentSessionId}` shape `SubAgentSessionStore` already tracks, Tier 3 always fails closed) — but **it has no production call site yet**. No verb today lets one live agent address another (`task` only spawns), so this rule is exercised only by `authority.test.ts` and is not enforced at runtime in this PR. It exists so CL-6942 (split spawn from wait) and CL-6944 (`send_input` steering) — the first verbs that make an agent addressable by another — can call it from day one instead of each inventing its own check. Treat it as unenforced until one of those wires a call site. -- `task()` is unaffected and remains the only spawn verb until the new verbs land beside it (deprecated-not-deleted per the CL-6940 epic). Its argument schema and wire contract are unchanged; the tier check only gates which packages may have it mounted at all. +- **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`, `interrupt_agent`, `close_agent`, `resume_agent`, `followup_task`, `read_agent_trace`, `search_agents`) plus reserved names (`list_agents`, `send_input`) so a later mount site inherits the same gate. +- **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. `interrupt_agent` terminalizes the wait mailbox immediately. +- `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/agent/directors/skywalker/package.test.ts b/src/agent/directors/skywalker/package.test.ts index c0a6e271f..10f871c75 100644 --- a/src/agent/directors/skywalker/package.test.ts +++ b/src/agent/directors/skywalker/package.test.ts @@ -102,6 +102,8 @@ describe("skywalkerPackage", () => { expect(p).toContain("wait_agents"); expect(p).toContain("Idle-orchestrator"); expect(p).toContain("deprecated fused spawn+wait"); + expect(p).toContain('mode="all"'); + expect(p).toContain("uncollected spawns"); expect(p).not.toContain("Present the plan when the change is large or ambiguous"); }); diff --git a/src/agent/directors/skywalker/package.ts b/src/agent/directors/skywalker/package.ts index 61acf8443..8c0668a00 100644 --- a/src/agent/directors/skywalker/package.ts +++ b/src/agent/directors/skywalker/package.ts @@ -16,7 +16,7 @@ You do not do the specialists' jobs by default. For tiny bounded product edits, Do not run long-blocking jobs on the parent (evals, full test suites, long installs, long-running implementation). Dispatch intern (mechanical shell), tester (suite / repro), or builder (substantial code). Path tools (write_file/edit_file/delete_file) are the DIY surface; shell file-writes stay denied. -Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents / task() right after spawn. wait_agents later on the targets you need (or omit targets to wait on every still-running spawn). task() still fuses spawn+wait and holds the parent until that one worker finishes. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents / task() holds those steers. A bare spawn_agent does not. +Idle-orchestrator: fire one or more spawn_agent calls in a turn — each returns immediately with an agent_id and does not hold the parent. Then **reply to the operator** with who is running and what happens next before you block. Prefer ending that turn (or calling wait_agents with a short timeout_ms) so Enter can land; do not immediately fuse into a long wait_agents / task() right after spawn. wait_agents later on the targets you need (or omit targets to wait on this session's own uncollected spawns — never a sibling's). Use mode="all" when you need every target to finish; interrupt_agent unblocks wait_agents immediately. A timeout means still running — do not tight-loop wait_agents hoping for a different answer. task() still fuses spawn+wait and holds the parent until that one worker finishes. Enter mid-run delivers at the next parent tool.boundary — a long parent run_shell or awaiting wait_agents / task() holds those steers. A bare spawn_agent does not. # Operator updates (mandatory while fleet is live) diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 23a46b14b..d25b312d6 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -346,7 +346,7 @@ export async function createAgentToolset(args: AgentToolsetArgs): Promise { expect(secondResults[0]!.report).toBe("finished"); }); - test("wait_agents with no targets waits on all currently running spawned agents", async () => { + test("wait_agents with no targets waits on all uncollected agents in this fleet", async () => { const gates = [deferred(), deferred()]; let callIndex = 0; const deps = makeDeps(async () => gates[callIndex++]!.promise); @@ -374,3 +375,258 @@ describe("fleetRecords retention cap", () => { expect(results[0]!.hint).toContain("read_agent_trace"); }); }); + +describe("spawn_agent parentage", () => { + test("records the caller session as parentSessionId", async () => { + const gate = deferred(); + const deps = makeDeps(async () => gate.promise); + deps.parentSessionId = "parent-orch"; + const spawn = createSpawnAgentTool(deps); + + const spawned = await callTool(spawn, { + description: "child", + prompt: "do it", + intent: "explore", + }); + const session = deps.sessions.get(spawned.agent_id as string); + expect(session?.parentSessionId).toBe("parent-orch"); + + gate.resolve({ report: "done" }); + }); +}); + +describe("wait_agents caller scope", () => { + test("omitted targets wait only on this fleet, not every running session in the shared store", async () => { + const gate = deferred(); + const deps = makeDeps(async () => gate.promise); + const foreign = deps.sessions.start({ + id: "foreign-sibling", + description: "someone else's worker", + agentId: "explorer", + brief: "b", + }); + deps.sessions.markRunning(foreign.id); + + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "mine", + prompt: "do it", + intent: "explore", + }); + + const waited = await callTool(wait, { timeout_ms: 50 }); + expect(waited.timed_out).toBe(true); + const results = waited.results as { agent_id: string; status: string }[]; + expect(results.map((r) => r.agent_id)).toEqual([spawned.agent_id as string]); + expect(results.every((r) => r.agent_id !== foreign.id)).toBe(true); + + gate.resolve({ report: "done" }); + }); + + test("mode=all stays blocked until every target is terminal", async () => { + const gates = [deferred(), deferred()]; + let callIndex = 0; + const deps = makeDeps(async () => gates[callIndex++]!.promise); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + + const first = await callTool(spawn, { + description: "a", + prompt: "do it", + intent: "explore", + }); + const second = await callTool(spawn, { + description: "b", + prompt: "do it", + intent: "explore", + }); + const ids = [first.agent_id as string, second.agent_id as string]; + + gates[0]!.resolve({ report: "a done" }); + const partial = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 50 }); + expect(partial.timed_out).toBe(true); + const partialResults = partial.results as { status: string }[]; + expect(partialResults.some((r) => r.status === "running")).toBe(true); + + gates[1]!.resolve({ report: "b done" }); + const finished = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 5000 }); + expect(finished.timed_out).toBe(false); + const finishedResults = finished.results as { status: string }[]; + expect(finishedResults.every((r) => r.status === "done")).toBe(true); + }); + + test("mode=all with one interrupted target stays blocked until siblings finish", async () => { + const gates = [deferred(), deferred()]; + let callIndex = 0; + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => {}, + interrupt: () => {}, + followup: async () => "", + }); + return gates[callIndex++]!.promise; + }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const interrupt = createInterruptAgentTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + + const first = await callTool(spawn, { + description: "a", + prompt: "do it", + intent: "explore", + }); + const second = await callTool(spawn, { + description: "b", + prompt: "do it", + intent: "explore", + }); + const ids = [first.agent_id as string, second.agent_id as string]; + + // Interrupt one of N before mode=all starts: interrupted is terminal for + // that target, but mode=all must not complete as "all done" while a + // sibling is still running. + if (interrupt.kind !== "full") throw new Error("expected full tool"); + await interrupt.handler( + { id: "int-1", name: "interrupt_agent", arguments: { target: ids[0]! } }, + new AbortController().signal, + ); + + const partial = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 50 }); + expect(partial.timed_out).toBe(true); + const partialResults = partial.results as { agent_id: string; status: string }[]; + expect(partialResults.find((r) => r.agent_id === ids[0]!)?.status).toBe("interrupted"); + expect(partialResults.find((r) => r.agent_id === ids[1]!)?.status).toBe("running"); + + gates[1]!.resolve({ report: "b done" }); + const finished = await callTool(wait, { targets: ids, mode: "all", timeout_ms: 5000 }); + expect(finished.timed_out).toBe(false); + const finishedResults = finished.results as { agent_id: string; status: string }[]; + expect(finishedResults.find((r) => r.agent_id === ids[0]!)?.status).toBe("interrupted"); + expect(finishedResults.find((r) => r.agent_id === ids[1]!)?.status).toBe("done"); + // Leave the interrupted gate unresolved — interrupt unblocked the wait + // without the run settling. + }); + + test("aborting the wait returns without cancelling workers", async () => { + const gate = deferred(); + const deps = makeDeps(async () => gate.promise); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const spawned = await callTool(spawn, { + description: "slow", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + + if (wait.kind !== "full") throw new Error("expected full tool"); + const ac = new AbortController(); + const started = Date.now(); + const pending = wait.handler( + { id: "wait-1", name: "wait_agents", arguments: { targets: [id], timeout_ms: 5000 } }, + ac.signal, + ); + ac.abort(); + const result = await pending; + expect(Date.now() - started).toBeLessThan(500); + const content = + typeof result.content === "string" ? result.content : JSON.stringify(result.content); + const parsed = JSON.parse(content) as { + timed_out: boolean; + results: { status: string }[]; + }; + expect(parsed.timed_out).toBe(true); + expect(parsed.results[0]!.status).toBe("running"); + expect(deps.sessions.get(id)?.status).toBe("running"); + + gate.resolve({ report: "done" }); + }); +}); + +describe("interrupt_agent unblocks wait_agents", () => { + test("interrupt marks the fleet record terminal so wait returns without the run settling", async () => { + const gate = deferred(); + const deps = makeDeps(async (params) => { + params.onAgentReady?.({ + close: async () => {}, + interrupt: () => {}, + followup: async () => "", + }); + return gate.promise; + }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + sessions: deps.sessions, + fleetRecords: deps.fleetRecords, + }); + const interrupt = createInterruptAgentTool({ + 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 }); + if (interrupt.kind !== "full") throw new Error("expected full tool"); + await interrupt.handler( + { id: "int-1", name: "interrupt_agent", arguments: { target: id } }, + new AbortController().signal, + ); + + const waited = await waiting; + expect(waited.timed_out).toBe(false); + const results = waited.results as { agent_id: string; status: string }[]; + expect(results).toEqual([{ agent_id: id, status: "interrupted" }]); + expect(deps.sessions.get(id)?.lifecycleStatus).toBe("interrupted"); + expect(deps.sessions.get(id)?.status).toBe("running"); + }); + + test("an interrupted run result terminalizes a still-running fleet record", async () => { + const settle = deferred(); + const deps = makeDeps(async () => settle.promise); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ + 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; + + settle.resolve({ + report: "## Summary\nStopped.\n## Findings\npartial\n## Blockers\ninterrupted\n## Paths\n", + interrupted: true, + }); + + const waited = await callTool(wait, { targets: [id], timeout_ms: 5000 }); + expect(waited.timed_out).toBe(false); + const results = waited.results as { status: string; report?: string }[]; + expect(results[0]!.status).toBe("interrupted"); + expect(results[0]!.report).toContain("partial"); + }); +}); diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 8ddf65dfc..c5a0940a6 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -2,10 +2,10 @@ * spawn_agent / wait_agents: the non-blocking half of fleet dispatch, * split out of `task()`'s fused spawn+wait. * - * `task()` (task-tool.ts) remains the fused, blocking primitive and is - * unchanged. These two verbs let an orchestrator start several workers in - * one turn (spawn_agent returns immediately) and later block on whichever - * ones it cares about (wait_agents), instead of one task() call per worker + * `task()` (task-tool.ts) remains the deprecated fused spawn+wait fallback. + * These two verbs are the supported fleet path: start several workers in one + * turn (spawn_agent returns immediately) and later block on this caller's + * own workers (wait_agents), instead of one task() call per worker * serializing the wait. * * Running state and the mailbox (`subscribe`) are the existing @@ -71,7 +71,7 @@ import { classifyAgentName } from "../telemetry/classify.js"; /** Terminal (or running) record for one spawned agent, keyed by agent id. */ interface FleetRecord { - status: "running" | "done" | "failed"; + status: "running" | "done" | "failed" | "interrupted"; report?: string; error?: string; /** Set once a wait_agents caller has been handed this result. */ @@ -96,19 +96,66 @@ export const MAX_FLEET_RECORDS = 200; */ class FleetRecords { private readonly records = new Map(); + private readonly listeners = new Set<() => void>(); + + subscribe(listener: () => void): () => void { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + + private notify(): void { + for (const listener of this.listeners) listener(); + } register(id: string): void { this.records.set(id, { status: "running" }); } resolve(id: string, report: string): void { + const existing = this.records.get(id); + if (existing !== undefined && existing.status !== "running") return; this.records.set(id, { status: "done", report }); this.enforceCap(); + this.notify(); } reject(id: string, error: string): void { + const existing = this.records.get(id); + if (existing !== undefined && existing.status !== "running") return; this.records.set(id, { status: "failed", error }); this.enforceCap(); + this.notify(); + } + + /** + * Marks a still-running record interrupted so wait_agents unblocks. + * No-op on an already-terminal id — interrupt must not clobber a collected + * report, and a late interrupt after complete/fail is meaningless. + */ + interrupt(id: string, report?: string): void { + const existing = this.records.get(id); + if (existing === undefined) return; + if (existing.status === "interrupted" && existing.collected !== true && report !== undefined) { + existing.report = report; + this.notify(); + return; + } + if (existing.status !== "running") return; + this.records.set(id, { + status: "interrupted", + ...(report !== undefined ? { report } : {}), + }); + this.enforceCap(); + this.notify(); + } + + /** Running plus terminal-but-not-yet-handed-to-a-waiter. */ + uncollectedIds(): string[] { + return [...this.records.entries()] + .filter(([, record]) => record.collected !== true) + .map(([id]) => id); } /** Read without consuming — used for the terminal-yet check. */ @@ -181,7 +228,7 @@ const SpawnAgentArgs = type({ export const spawnAgentToolDefinition: ToolDefinition = { name: "spawn_agent", description: - "Start a worker agent and return IMMEDIATELY with its agent_id — this never blocks on the worker's completion. Same brief fields as task() (description/prompt/context/goals/intent/success_criteria/do_not/report_focus); pass agent= a director id or intent= (one of explore|implement|review|plan|general). Fire several spawn_agent calls in one turn to start workers in parallel, then use wait_agents to block on whichever ones you need next. Prefer task() when you only need one worker and want its result before doing anything else — spawn_agent+wait_agents earns its keep when you want to start more than one worker without stalling on the first.", + "Start a worker agent and return IMMEDIATELY with its agent_id — this never blocks on the worker's completion. Same brief fields as task() (description/prompt/context/goals/intent/success_criteria/do_not/report_focus); pass agent= a director id or intent= (one of explore|implement|review|plan|general). Fire several spawn_agent calls in one turn to start workers in parallel, then use wait_agents to collect them. task() is the deprecated fused spawn+wait fallback for a single blocking worker.", inputSchema: { type: "object", properties: { @@ -221,6 +268,7 @@ export const spawnAgentToolDefinition: ToolDefinition = { const WaitAgentsArgs = type({ "targets?": "string[]", "timeout_ms?": "number", + "mode?": "'any' | 'all'", }); export const DEFAULT_WAIT_TIMEOUT_MS = 30_000; @@ -229,13 +277,15 @@ export const MAX_WAIT_TIMEOUT_MS = 300_000; export const waitAgentsToolDefinition: ToolDefinition = { name: "wait_agents", description: - `Block until any of the given agents (default: every agent you have spawned that is still running) reaches a ` + - `terminal state, or timeout_ms elapses — whichever comes first. Default timeout ${DEFAULT_WAIT_TIMEOUT_MS}ms, ` + - `clamped to a ${MAX_WAIT_TIMEOUT_MS}ms max. A timeout is NOT an error and never touches the workers — they keep ` + - `running exactly as before and remain waitable. Do not call this in a tight zero-progress loop hoping for a ` + - `different answer: a timeout means "still running", not "try again right away" — either do other useful work ` + - `first, or call again with a longer timeout_ms. Calling again immediately with the same targets is safe (it is ` + - `a real timed wait, not a spin) but wastes turns if nothing about the situation has changed.`, + `Block until the given agents reach a terminal state (done, failed, or interrupted), or timeout_ms elapses. ` + + `Default mode is "any" (return when the first target finishes). Pass mode="all" to wait until every target is ` + + `terminal. Omit targets to wait on this caller's own uncollected fleet — the workers this spawn_agent/` + + `wait_agents pair started — never every running session in the shared store. Default timeout ${DEFAULT_WAIT_TIMEOUT_MS}ms, ` + + `clamped to a ${MAX_WAIT_TIMEOUT_MS}ms max. A timeout or parent-turn abort is NOT an error and never touches ` + + `the workers — they keep running and remain waitable. interrupt_agent unblocks this wait immediately with ` + + `status "interrupted". Do not call this in a tight zero-progress loop: a timeout means "still running", not ` + + `"try again right away" — do other work, reply to the operator, or change the brief. Calling again with the ` + + `same targets is a real timed wait, not a spin, but wastes turns if nothing has changed.`, inputSchema: { type: "object", properties: { @@ -243,12 +293,18 @@ export const waitAgentsToolDefinition: ToolDefinition = { type: "array", items: { type: "string" }, description: - "agent_id values to wait on. Omit to wait on every currently-running spawned agent.", + "agent_id values to wait on. Omit to wait on this caller's uncollected spawned agents only.", }, timeout_ms: { type: "number", description: `Max time to block, in ms. Default ${DEFAULT_WAIT_TIMEOUT_MS}, clamped to ${MAX_WAIT_TIMEOUT_MS}.`, }, + mode: { + type: "string", + enum: ["any", "all"], + description: + '"any" (default) returns when the first target is terminal. "all" waits until every target is terminal.', + }, }, }, }; @@ -260,6 +316,12 @@ export type AgentFleetDeps = SubAgentSandboxDeps & { run: (params: RunSubAgentParams) => Promise; sessions: SubAgentSessionStore; fleetRecords: FleetRecordsHandle; + /** + * Session id of the caller that is mounting this spawn_agent. Nested + * orchestrators pass their own worker id so close_agent can walk the tree. + * Omit on the primary session — its children are top-level. + */ + parentSessionId?: string; settings?: Settings | (() => Settings | undefined); catalog?: readonly ProviderCatalogEntry[] | (() => readonly ProviderCatalogEntry[]); onEvent?: (event: ReactorEmittedEvent) => void; @@ -399,6 +461,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { // of being torn down — close_agent (or resume_agent, transitively) // governs it from here on. retained: true, + ...(deps.parentSessionId !== undefined ? { parentSessionId: deps.parentSessionId } : {}), }); deps.fleetRecords.register(session.id); const agentName = classifyAgentName(resolved.directorId); @@ -470,8 +533,12 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { // interrupt_agent already flipped this session to "interrupted" // synchronously (session-store.interruptOne) — do not let the // settling promise's normal bookkeeping overwrite that with a - // "completed" status. - if (result.interrupted === true) return; + // "completed" status. Still terminalize fleetRecords so a waiter + // that never saw interrupt_agent (or raced it) cannot hang. + if (result.interrupted === true) { + deps.fleetRecords.interrupt(session.id, result.report); + return; + } // Operator cancel may race after run resolves (childCtl aborted). // Keep strip status cancelled when sessions.cancel already flipped // it, but never discard a returned body (including salvage) — @@ -514,26 +581,50 @@ interface WaitAgentsDeps { fleetRecords: FleetRecordsHandle; } +function isSoftInterrupted( + session: ReturnType, +): session is NonNullable> { + // interrupt_agent keeps strip status "running" so followup_task can reuse + // the session. cancel() also sets lifecycleStatus "interrupted" but flips + // status to "cancelled" — that path still owes wait_agents a salvage + // report via fleetRecords, so it is not wait-terminal on its own. + return ( + session !== undefined && + session.status === "running" && + (session.lifecycleStatus === "interrupted" || session.lifecycleStatus === "shutdown") + ); +} + +function isWaitTerminal( + id: string, + sessions: SubAgentSessionStore, + fleetRecords: FleetRecordsHandle, +): boolean { + const record = fleetRecords.peek(id); + if (record !== undefined && record.status !== "running") return true; + return isSoftInterrupted(sessions.get(id)); +} + /** - * Blocks until any of `targets` is terminal in `fleetRecords`, or `timeoutMs` - * elapses. Driven by the session store's mailbox (`subscribe`) — complete()/ - * fail() always write fleetRecords before notifying, so a synchronous - * subscriber sees the up-to-date record — raced against a timer; never - * polls. Returns without side effects on timeout: nothing is touched, so a - * caller can wait again immediately. + * Blocks until `mode` is satisfied for `targets`, or `timeoutMs` / abort + * elapses. Driven by the session store's mailbox (`subscribe`) raced against + * a timer and the parent tool signal; never polls. Timeout and abort have no + * side effects: workers keep running and remain waitable. */ -async function waitForAnyTerminal( +async function waitForTerminal( sessions: SubAgentSessionStore, fleetRecords: FleetRecordsHandle, targets: readonly string[], timeoutMs: number, + mode: "any" | "all", + signal?: AbortSignal, ): Promise { - const anyTerminal = (): boolean => - targets.some((id) => { - const record = fleetRecords.peek(id); - return record !== undefined && record.status !== "running"; - }); - if (anyTerminal()) return false; + const ready = (): boolean => + mode === "all" + ? targets.every((id) => isWaitTerminal(id, sessions, fleetRecords)) + : targets.some((id) => isWaitTerminal(id, sessions, fleetRecords)); + if (signal?.aborted) return true; + if (ready()) return false; return await new Promise((resolve) => { let settled = false; @@ -541,64 +632,80 @@ async function waitForAnyTerminal( if (settled) return; settled = true; clearTimeout(timer); - unsubscribe(); + unsubscribeSessions(); + unsubscribeFleet(); + signal?.removeEventListener("abort", onAbort); resolve(timedOut); }; + const onAbort = (): void => finish(true); + const onChange = (): void => { + if (ready()) finish(false); + }; const timer = setTimeout(() => finish(true), timeoutMs); - const unsubscribe = sessions.subscribe(() => { - if (anyTerminal()) finish(false); - }); + const unsubscribeSessions = sessions.subscribe(onChange); + const unsubscribeFleet = fleetRecords.subscribe(onChange); + signal?.addEventListener("abort", onAbort, { once: true }); + if (signal?.aborted) finish(true); }); } export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { return tool({ definition: waitAgentsToolDefinition, - handler: async (call, _signal): Promise => { + handler: async (call, signal): Promise => { const parsed = WaitAgentsArgs(call.arguments); if (parsed instanceof type.errors) { return fleetResult(call.id, `Error: wait_agents arguments invalid: ${parsed.summary}`); } const requestedTimeout = parsed.timeout_ms ?? DEFAULT_WAIT_TIMEOUT_MS; const timeoutMs = Math.min(Math.max(requestedTimeout, 0), MAX_WAIT_TIMEOUT_MS); + const mode = parsed.mode ?? "any"; const targets = parsed.targets !== undefined && parsed.targets.length > 0 ? parsed.targets - : deps.sessions - .list() - .filter((s) => s.status === "running") - .map((s) => s.id); + : deps.fleetRecords.uncollectedIds(); if (targets.length === 0) { return fleetResult(call.id, JSON.stringify({ results: [], timed_out: false })); } - const timedOut = await waitForAnyTerminal( + const timedOut = await waitForTerminal( deps.sessions, deps.fleetRecords, targets, timeoutMs, + mode, + signal, ); - // Terminal records are consumed (removed) once delivered here; a - // running record is only peeked, so it stays waitable. + // Terminal fleet records are marked collected once delivered here; a + // running record is only peeked, so it stays waitable. Session + // lifecycle is a fallback for interrupt/close that raced the record. const results = targets.map((id) => { const record = deps.fleetRecords.peek(id); + if (record !== undefined && record.status !== "running") { + const taken = deps.fleetRecords.take(id) ?? record; + return { + agent_id: id, + status: taken.status, + ...(taken.report !== undefined ? { report: taken.report } : {}), + ...(taken.error !== undefined ? { error: taken.error } : {}), + ...(taken.hint !== undefined ? { hint: taken.hint } : {}), + }; + } + const session = deps.sessions.get(id); + if (isSoftInterrupted(session)) { + return { + agent_id: id, + status: "interrupted" as const, + ...(session.report !== undefined ? { report: session.report } : {}), + }; + } if (record === undefined) { return { agent_id: id, status: "unknown" as const }; } - if (record.status === "running") { - return { agent_id: id, status: "running" as const }; - } - const taken = deps.fleetRecords.take(id) ?? record; - return { - agent_id: id, - status: taken.status, - ...(taken.report !== undefined ? { report: taken.report } : {}), - ...(taken.error !== undefined ? { error: taken.error } : {}), - ...(taken.hint !== undefined ? { hint: taken.hint } : {}), - }; + return { agent_id: id, status: "running" as const }; }); return fleetResult(call.id, JSON.stringify({ results, timed_out: timedOut })); diff --git a/src/subagent/authority.ts b/src/subagent/authority.ts index 3a042eb52..10bfe9e25 100644 --- a/src/subagent/authority.ts +++ b/src/subagent/authority.ts @@ -5,12 +5,12 @@ * in a prompt. This module owns two checks: * * - assertTierMayMountFleetVerb: a Tier 3 leaf may never mount a fleet verb - * (today: task, search_agents, read_agent_trace; the spawn_agent/ - * wait_agents/list_agents/send_input/interrupt_agent/close_agent/ - * resume_agent/followup_task verbs land in later child issues against - * this same gate). 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. + * (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. * - 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,9 +25,9 @@ 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. Verbs not yet - * implemented are listed here so their eventual mount sites inherit the gate - * for free instead of needing a second allowlist. + * 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. */ export const FLEET_VERBS = new Set([ "task", @@ -109,14 +109,9 @@ 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. + * 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` diff --git a/src/subagent/lifecycle-tools.test.ts b/src/subagent/lifecycle-tools.test.ts index a853022d1..8d57eec77 100644 --- a/src/subagent/lifecycle-tools.test.ts +++ b/src/subagent/lifecycle-tools.test.ts @@ -6,6 +6,7 @@ import { createInterruptAgentTool, createFollowupTaskTool, } from "./lifecycle-tools.js"; +import { createFleetRecords } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; async function callTool( @@ -135,7 +136,10 @@ describe("interrupt_agent / followup_task", () => { return `Applying fix given ${history.length} prior turns of context.`; }); - const interruptAgent = createInterruptAgentTool({ sessions }); + const interruptAgent = createInterruptAgentTool({ + sessions, + fleetRecords: createFleetRecords(), + }); const followupTask = createFollowupTaskTool({ sessions }); const interruptResult = await callTool(interruptAgent, { target: worker.id }); @@ -219,7 +223,10 @@ describe("interrupt_agent / followup_task", () => { }); sessions.registerFollowup(worker.id, async () => "resumed cleanly"); - const interruptAgent = createInterruptAgentTool({ sessions }); + const interruptAgent = createInterruptAgentTool({ + sessions, + fleetRecords: createFleetRecords(), + }); const followupTask = createFollowupTaskTool({ sessions }); await callTool(interruptAgent, { target: worker.id }); @@ -239,7 +246,10 @@ describe("interrupt_agent / followup_task", () => { const notRunning = sessions.start({ description: "d", agentId: "a", brief: "b" }); sessions.complete(notRunning.id, "## Summary\nDone."); - const interruptAgent = createInterruptAgentTool({ sessions }); + const interruptAgent = createInterruptAgentTool({ + sessions, + fleetRecords: createFleetRecords(), + }); const followupTask = createFollowupTaskTool({ sessions }); if (interruptAgent.kind !== "full") throw new Error("expected full tool"); diff --git a/src/subagent/lifecycle-tools.ts b/src/subagent/lifecycle-tools.ts index 68601b556..180f6cbab 100644 --- a/src/subagent/lifecycle-tools.ts +++ b/src/subagent/lifecycle-tools.ts @@ -17,6 +17,7 @@ import { type } from "arktype"; 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"; function lifecycleResult(callId: string, content: string): ToolResult { @@ -86,8 +87,15 @@ function descendantsClosingOrder( export interface LifecycleToolDeps { sessions: SubAgentSessionStore; + /** Optional for close/resume/followup; interrupt requires it (see InterruptAgentToolDeps). */ + fleetRecords?: FleetRecordsHandle; } +/** interrupt_agent always terminalizes the wait mailbox — no silent skip. */ +export type InterruptAgentToolDeps = LifecycleToolDeps & { + fleetRecords: FleetRecordsHandle; +}; + export function createCloseAgentTool(deps: LifecycleToolDeps): AgentTool { return tool({ definition: closeAgentToolDefinition, @@ -155,7 +163,8 @@ export const interruptAgentToolDefinition: ToolDefinition = { name: "interrupt_agent", description: "Stop a worker session's current turn while keeping the session and its context intact and " + - "reusable — distinct from close_agent, which is permanent. The worker's in-flight tool call or " + + "reusable — distinct from close_agent, which is permanent. Unblocks any in-flight wait_agents " + + "on this id immediately with status 'interrupted'. The worker's in-flight tool call or " + "inference keeps running in the background (there is no way to hard-stop it without tearing the " + "session down); this only stops the caller from waiting on it and marks the session " + "'interrupted' so followup_task or resume_agent can pick it back up with full prior context. " + @@ -169,7 +178,7 @@ export const interruptAgentToolDefinition: ToolDefinition = { }, }; -export function createInterruptAgentTool(deps: LifecycleToolDeps): AgentTool { +export function createInterruptAgentTool(deps: InterruptAgentToolDeps): AgentTool { return tool({ definition: interruptAgentToolDefinition, handler: async (call, _signal): Promise => { @@ -188,6 +197,9 @@ export function createInterruptAgentTool(deps: LifecycleToolDeps): AgentTool { `Error: cannot interrupt "${target}" (status: ${outcome.status}).`, ); } + // Wait mailbox is separate from the TUI strip — flip it here so + // wait_agents does not stay blocked on a still-"running" record. + deps.fleetRecords.interrupt(target); return lifecycleResult( call.id, JSON.stringify({ agent_id: target, status: "interrupted" satisfies AgentLifecycleStatus }), diff --git a/src/subagent/run-authority.test.ts b/src/subagent/run-authority.test.ts index 3d674bc7c..182067e5d 100644 --- a/src/subagent/run-authority.test.ts +++ b/src/subagent/run-authority.test.ts @@ -167,3 +167,49 @@ describe("runSubAgent search_agents mount gate (CL-7051, Tier-1 only)", () => { expect(searchAgentsMounts).toBe(1); }); }); + +describe("runSubAgent passes parentSessionId into spawn_agent mount", () => { + test("nested orchestrator fleetDeps.parentSessionId equals params.id", async () => { + const cwd = await tmpCwd(); + let capturedParentSessionId: string | undefined; + let spawnMounts = 0; + + await withMockedModuleDuring( + import.meta.resolve("./agent-fleet.js"), + (real: typeof import("./agent-fleet.js")) => ({ + ...real, + createSpawnAgentTool: (deps: Parameters[0]) => { + spawnMounts++; + capturedParentSessionId = deps.parentSessionId; + return real.createSpawnAgentTool(deps); + }, + }), + async () => { + const { runSubAgent: run } = await import("./run.js"); + try { + await run({ + ...baseParams(cwd, join(cwd, ".ctx")), + id: "greybeard-session", + orchestrator: true, + orchestratorTier: "nested-orchestrator", + nestedDispatch: { + permissionGate: testPermissionGate, + getWorkdirBase: () => join(cwd, ".ctx"), + provider: { + providerName: "test", + baseURL: "http://localhost", + model: "test-model", + }, + profiles: [{ id: "intern", systemPromptRole: "You are intern." }], + }, + }); + } catch { + // Inference/agent construction may fail; mount decisions run first. + } + }, + ); + + expect(spawnMounts).toBe(1); + expect(capturedParentSessionId).toBe("greybeard-session"); + }); +}); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index a5b4544e0..f71312827 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -585,6 +585,7 @@ export async function runSubAgent(params: RunSubAgentParams): Promise