From c044a60ebfe71eb5d021912e1ad11d6e47443adb Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 23 Aug 2026 22:21:33 -0700 Subject: [PATCH 1/4] Split task() into non-blocking spawn_agent + wait_agents spawn_agent starts a worker and returns immediately with {agent_id, status}; wait_agents blocks on any of a target set (default: all live agents) reaching a terminal state, or a clamped timeout, without touching the workers on timeout. task() is unchanged. --- CHANGELOG.md | 11 + src/subagent/agent-fleet.test.ts | 172 ++++++++++++ src/subagent/agent-fleet.ts | 455 +++++++++++++++++++++++++++++++ src/subagent/run.ts | 37 ++- 4 files changed, 674 insertions(+), 1 deletion(-) create mode 100644 src/subagent/agent-fleet.test.ts create mode 100644 src/subagent/agent-fleet.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c1d84ca..30f34d65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,17 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename - Tier 3 leaf workers can now report via `submit_result`, a typed channel alongside the markdown envelope that validates against a director-declared JSON Schema and returns a correction (capped at 3 rounds) on an invalid submission. +- **`spawn_agent` / `wait_agents` split the fused spawn+wait out of `task()`.** + `spawn_agent` starts a worker and returns immediately with `{ agent_id, + status: "running" }` — it never awaits the worker's completion. `wait_agents` + blocks until any of the given (or, if omitted, all currently running) + agent ids reaches a terminal state, or `timeout_ms` elapses (default + 30s, clamped to a 300s max); a timeout is not an error and never touches + the workers — they keep running and stay waitable. Lets an orchestrator + fire several workers in one turn instead of serializing one `task()` call + per worker. `task()` is unchanged and remains the single-call spawn+block + primitive for the common one-worker case. + - **Fleet authority tiers are now runtime-enforced, not documented in a prompt.** Every director package carries a required `tier` (`orchestrator` / `nested-orchestrator` / `leaf`): skywalker gets full fleet control, greybeard diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts new file mode 100644 index 00000000..ac43c517 --- /dev/null +++ b/src/subagent/agent-fleet.test.ts @@ -0,0 +1,172 @@ +import { describe, expect, test } from "bun:test"; + +import { createSpawnAgentTool, createWaitAgentsTool, type AgentFleetDeps } from "./agent-fleet.js"; +import { createSubAgentSessionStore } from "./session-store.js"; +import { createPermissionGate } from "../permission/gate.js"; +import type { RunSubAgentParams } from "./types.js"; + +const testPermissionGate = createPermissionGate({ + approvals: [], + interactive: false, + skipPermissions: true, +}); + +const provider = { + providerName: "test-provider", + baseURL: "http://localhost", + model: "test-model", +}; + +function deferred(): { + promise: Promise; + resolve: (v: T) => void; + reject: (e: unknown) => void; +} { + let resolve!: (v: T) => void; + let reject!: (e: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function makeDeps(run: (params: RunSubAgentParams) => Promise): AgentFleetDeps { + return { + permissionGate: testPermissionGate, + cwd: "/tmp", + getWorkdirBase: () => "/tmp/workdir", + provider, + run, + sessions: createSubAgentSessionStore(), + }; +} + +async function callTool( + tool: ReturnType | ReturnType, + args: Record, +): Promise> { + if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`); + const result = await tool.handler( + { id: `call-${Math.random()}`, name: tool.definition.name, arguments: args }, + new AbortController().signal, + ); + const content = + typeof result.content === "string" ? result.content : JSON.stringify(result.content); + return JSON.parse(content); +} + +describe("spawn_agent", () => { + test("returns immediately with a running agent_id without waiting for the worker", async () => { + const gate = deferred(); + const deps = makeDeps(async () => gate.promise); + const spawn = createSpawnAgentTool(deps); + + const started = Date.now(); + const result = await callTool(spawn, { + description: "job", + prompt: "do it", + intent: "explore", + }); + const elapsed = Date.now() - started; + + expect(result.status).toBe("running"); + expect(typeof result.agent_id).toBe("string"); + expect(elapsed).toBeLessThan(1000); + + // Worker is still pending; store confirms it has not finished. + expect(deps.sessions.get(result.agent_id as string)?.status).toBe("running"); + + gate.resolve("done"); + }); +}); + +describe("spawn_agent + wait_agents", () => { + test("wait_agents on one target returns once it completes while siblings keep running", async () => { + const gates = [deferred(), deferred(), deferred()]; + let callIndex = 0; + const deps = makeDeps(async () => { + const i = callIndex++; + return gates[i]!.promise; + }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ sessions: deps.sessions }); + + const spawned = await Promise.all( + [0, 1, 2].map((i) => + callTool(spawn, { description: `job-${i}`, prompt: "do it", intent: "explore" }), + ), + ); + const ids = spawned.map((s) => s.agent_id as string); + + gates[0]!.resolve("first report"); + + const waited = await callTool(wait, { targets: [ids[0]], timeout_ms: 5000 }); + expect(waited.timed_out).toBe(false); + const results = waited.results as { agent_id: string; status: string; report?: string }[]; + expect(results).toHaveLength(1); + expect(results[0]!.status).toBe("done"); + expect(results[0]!.report).toBe("first report"); + + // The other two remain untouched and running. + expect(deps.sessions.get(ids[1]!)?.status).toBe("running"); + expect(deps.sessions.get(ids[2]!)?.status).toBe("running"); + + gates[1]!.resolve("second"); + gates[2]!.resolve("third"); + }); + + test("wait_agents times out on a still-running agent without cancelling it, and can be called again", async () => { + const gate = deferred(); + const deps = makeDeps(async () => gate.promise); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ sessions: deps.sessions }); + + const spawned = await callTool(spawn, { + description: "slow job", + prompt: "do it", + intent: "explore", + }); + const id = spawned.agent_id as string; + + const first = await callTool(wait, { targets: [id], timeout_ms: 50 }); + expect(first.timed_out).toBe(true); + const firstResults = first.results as { agent_id: string; status: string }[]; + expect(firstResults[0]!.status).toBe("running"); + + // Not cancelled, not failed — still running. + expect(deps.sessions.get(id)?.status).toBe("running"); + + // A second wait still works cleanly (either another timeout, or completion). + gate.resolve("finished"); + const second = await callTool(wait, { targets: [id], timeout_ms: 5000 }); + expect(second.timed_out).toBe(false); + const secondResults = second.results as { + agent_id: string; + status: string; + report?: string; + }[]; + expect(secondResults[0]!.status).toBe("done"); + expect(secondResults[0]!.report).toBe("finished"); + }); + + test("wait_agents with no targets waits on all currently running spawned agents", 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 }); + + await callTool(spawn, { description: "a", prompt: "do it", intent: "explore" }); + await callTool(spawn, { description: "b", prompt: "do it", intent: "explore" }); + + gates[0]!.resolve("a done"); + const result = await callTool(wait, { timeout_ms: 5000 }); + expect(result.timed_out).toBe(false); + const results = result.results as { status: string }[]; + expect(results).toHaveLength(2); + expect(results.some((r) => r.status === "done")).toBe(true); + + gates[1]!.resolve("b done"); + }); +}); diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts new file mode 100644 index 00000000..5e448200 --- /dev/null +++ b/src/subagent/agent-fleet.ts @@ -0,0 +1,455 @@ +/** + * spawn_agent / wait_agents (CL-6942): 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 + * serializing the wait. + * + * State lives entirely in the existing SubAgentSessionStore (status/report/ + * error, keyed by session id) — no parallel bookkeeping map. wait_agents' + * blocking is driven by the store's `subscribe` mailbox raced against a + * timeout timer; it never polls. + * + * Argument shape intentionally mirrors `task()`'s (description/prompt/ + * context/goals/intent/success_criteria/do_not/report_focus/maxTurns) so a + * caller can swap one for the other. Scope is deliberately narrower than + * `task()` for this first cut: only closed-director dispatch (`agent=` a + * director id, or `intent=`) is supported — no custom AgentProfile lookup, + * no worktree isolation, no nested orchestration, no re-dispatch ledger. + * Those are `task()`-only for now; nothing here stops adding them later. + */ + +import { tool } from "@intx/agent"; +import type { AgentTool } from "@intx/agent"; +import { type } from "arktype"; +import type { ToolDefinition, ToolResult } from "@intx/types/runtime"; +import type { ReactorEmittedEvent } from "@intx/inference"; + +import type { ProviderCatalogEntry } from "../config/index.js"; +import { + isDirectorId, + packageToCapabilities, + resolveDirector, +} from "../agent/directors/registry.js"; +import { + defaultEffortForDirector, + formatDirectorSystemPrompt, +} from "../agent/directors/identity.js"; +import type { Settings } from "../config/settings.js"; +import { resolveSubAgentMaxTurns, validateTaskMaxTurns } from "../config/settings.js"; +import { resolveEffortForRole } from "../provider/reasoning-effort.js"; +import { isCodexProviderName } from "../config/codex-providers.js"; +import { buildDispatchBrief, type TaskIntent } from "./report.js"; +import type { SubAgentSessionStore, SubAgentSessionStatus } from "./session-store.js"; +import type { RunSubAgentParams, SubAgentProvider, SubAgentSandboxDeps } from "./types.js"; +import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js"; +import { classifyAgentName } from "../telemetry/classify.js"; + +const SpawnAgentArgs = type({ + description: "string", + prompt: "string", + "context?": "string", + "agent?": "string", + "goals?": "string[]", + "intent?": "'explore' | 'implement' | 'review' | 'plan' | 'general'", + "success_criteria?": "string[]", + "do_not?": "string[]", + "report_focus?": "string", + "maxTurns?": "number", +}); + +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/maxTurns); 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.", + inputSchema: { + type: "object", + properties: { + description: { type: "string", description: "A short label for the worker job." }, + prompt: { type: "string", description: "The actionable goal for the worker." }, + context: { type: "string", description: "Optional durable background." }, + goals: { + type: "array", + items: { type: "string" }, + description: "Optional ordered checklist seeds for the worker's own manage_tasks list.", + }, + intent: { + type: "string", + enum: ["explore", "implement", "review", "plan", "general"], + description: "Optional spawn intent; selects a closed director when agent= is omitted.", + }, + success_criteria: { + type: "array", + items: { type: "string" }, + description: "Optional concrete done checks.", + }, + do_not: { + type: "array", + items: { type: "string" }, + description: "Optional explicit out-of-scope actions.", + }, + report_focus: { type: "string", description: "Optional hint for what Findings must cover." }, + agent: { + type: "string", + description: "Optional director id (e.g. from search_agents). Alternative to intent=.", + }, + maxTurns: { + type: "number", + description: "Optional inference-turn budget for this worker only.", + }, + }, + required: ["description", "prompt"], + }, +}; + +const WaitAgentsArgs = type({ + "targets?": "string[]", + "timeout_ms?": "number", +}); + +export const DEFAULT_WAIT_TIMEOUT_MS = 30_000; +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.`, + inputSchema: { + type: "object", + properties: { + targets: { + type: "array", + items: { type: "string" }, + description: + "agent_id values to wait on. Omit to wait on every currently-running spawned agent.", + }, + timeout_ms: { + type: "number", + description: `Max time to block, in ms. Default ${DEFAULT_WAIT_TIMEOUT_MS}, clamped to ${MAX_WAIT_TIMEOUT_MS}.`, + }, + }, + }, +}; + +export type AgentFleetDeps = SubAgentSandboxDeps & { + cwd: string; + getWorkdirBase: () => string; + provider: SubAgentProvider | (() => SubAgentProvider); + run: (params: RunSubAgentParams) => Promise; + sessions: SubAgentSessionStore; + settings?: Settings | (() => Settings | undefined); + catalog?: readonly ProviderCatalogEntry[] | (() => readonly ProviderCatalogEntry[]); + onEvent?: (event: ReactorEmittedEvent) => void; + onProgress?: (info: { description: string; toolName: string }) => void; + telemetry?: Telemetry; +}; + +function resolveDep(value: T | (() => T)): T { + return typeof value === "function" ? (value as () => T)() : value; +} + +function fleetResult(callId: string, content: string): ToolResult { + const isError = content.startsWith("Error:"); + return { callId, content, ...(isError ? { isError: true } : {}) }; +} + +/** Resolve agent=/intent= to a closed director. Mirrors task()'s director-only branch. */ +function resolveDirectorDispatch( + agentId: string | undefined, + intent: TaskIntent | undefined, +): + | { + ok: true; + directorId: string; + systemPromptRole: string; + capabilities: ReturnType; + roleDefault: ReturnType; + } + | { ok: false; error: string } { + if (agentId !== undefined && agentId.length > 0) { + if (!isDirectorId(agentId)) { + return { + ok: false, + error: `Error: unknown director "${agentId}". spawn_agent only supports closed director ids (call search_agents to discover them) or intent=.`, + }; + } + const resolved = resolveDirector({ agentId }); + if (!resolved.ok) return { ok: false, error: `Error: ${resolved.error} ${resolved.hint}` }; + const pkg = resolved.package; + return { + ok: true, + directorId: pkg.id, + systemPromptRole: formatDirectorSystemPrompt(pkg), + capabilities: packageToCapabilities(pkg), + roleDefault: defaultEffortForDirector(pkg), + }; + } + if (intent !== undefined) { + const resolved = resolveDirector({ intent }); + if (!resolved.ok) return { ok: false, error: `Error: ${resolved.error} ${resolved.hint}` }; + const pkg = resolved.package; + return { + ok: true, + directorId: pkg.id, + systemPromptRole: formatDirectorSystemPrompt(pkg), + capabilities: packageToCapabilities(pkg), + roleDefault: defaultEffortForDirector(pkg), + }; + } + return { + ok: false, + error: + "Error: No director selected. Pass spawn_agent(agent=…) for a named director, or spawn_agent(intent=implement|explore|plan|review).", + }; +} + +export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { + const telemetry = deps.telemetry ?? NOOP_TELEMETRY; + return tool({ + definition: spawnAgentToolDefinition, + handler: async (call, _signal): Promise => { + const args = call.arguments; + const parsed = SpawnAgentArgs(args); + if (parsed instanceof type.errors) { + return fleetResult(call.id, `Error: spawn_agent arguments invalid: ${parsed.summary}`); + } + const { + description: rawDesc, + context: rawCtx, + prompt: rawPrompt, + agent: agentId, + goals: rawGoals, + intent: rawIntent, + success_criteria: rawSuccessCriteria, + do_not: rawDoNot, + report_focus: rawReportFocus, + maxTurns: rawMaxTurns, + } = parsed; + const description = rawDesc.trim(); + const prompt = rawPrompt.trim(); + if (description.length === 0 || prompt.length === 0) { + return fleetResult( + call.id, + "Error: spawn_agent requires a non-empty description and prompt.", + ); + } + const context = rawCtx?.trim(); + const goals = rawGoals?.map((g) => g.trim()).filter((g) => g.length > 0) ?? []; + const intent = rawIntent as TaskIntent | undefined; + const successCriteria = + rawSuccessCriteria?.map((c) => c.trim()).filter((c) => c.length > 0) ?? []; + const doNot = rawDoNot?.map((d) => d.trim()).filter((d) => d.length > 0) ?? []; + const reportFocus = rawReportFocus?.trim(); + + const resolved = resolveDirectorDispatch(agentId, intent); + if (!resolved.ok) return fleetResult(call.id, resolved.error); + + let taskMaxTurns: number | undefined; + if (rawMaxTurns !== undefined) { + const verdict = validateTaskMaxTurns(rawMaxTurns); + if (!verdict.ok) return fleetResult(call.id, `Error: ${verdict.message}`); + taskMaxTurns = verdict.value; + } + const settings = deps.settings !== undefined ? resolveDep(deps.settings) : undefined; + const resolvedMaxTurns = resolveSubAgentMaxTurns({ + ...(settings !== undefined ? { settings } : {}), + ...(taskMaxTurns !== undefined ? { taskMaxTurns } : {}), + }); + + let provider: SubAgentProvider = resolveDep(deps.provider); + const effort = resolveEffortForRole({ + orchestrator: false, + roleDefault: resolved.roleDefault, + ...(provider.reasoningEffort !== undefined + ? { parentEffort: provider.reasoningEffort } + : {}), + model: provider.model, + isCodex: isCodexProviderName(provider.providerName), + }); + provider = effort !== undefined ? { ...provider, reasoningEffort: effort } : provider; + + const brief = buildDispatchBrief({ + description, + prompt, + ...(context !== undefined && context.length > 0 ? { context } : {}), + ...(goals.length > 0 ? { goals } : {}), + ...(intent !== undefined ? { intent } : {}), + ...(successCriteria.length > 0 ? { successCriteria } : {}), + ...(doNot.length > 0 ? { doNot } : {}), + ...(reportFocus !== undefined && reportFocus.length > 0 ? { reportFocus } : {}), + }); + + const session = deps.sessions.start({ + description, + agentId: resolved.directorId, + brief, + }); + const agentName = classifyAgentName(resolved.directorId); + telemetry.capture("subagent_start", { agent_name: agentName }); + const startedAt = Date.now(); + + const childCtl = new AbortController(); + deps.sessions.registerCancel(session.id, () => { + if (!childCtl.signal.aborted) childCtl.abort(); + }); + + const onEvent = (event: ReactorEmittedEvent): void => { + deps.sessions.appendEvent(session.id, event); + deps.onEvent?.(event); + }; + + const catalog = deps.catalog !== undefined ? resolveDep(deps.catalog) : undefined; + const params: RunSubAgentParams = { + permissionGate: deps.permissionGate, + ...(deps.inheritMcpTools !== undefined ? { inheritMcpTools: deps.inheritMcpTools } : {}), + ...(deps.shellTimeout !== undefined ? { shellTimeout: deps.shellTimeout } : {}), + ...(deps.shellEnv !== undefined ? { shellEnv: deps.shellEnv } : {}), + ...(deps.extraToolPlugins !== undefined ? { extraToolPlugins: deps.extraToolPlugins } : {}), + ...(deps.getBlobReader !== undefined ? { getBlobReader: deps.getBlobReader } : {}), + cwd: deps.cwd, + workdirBase: deps.getWorkdirBase(), + provider, + ...(settings !== undefined ? { settings } : {}), + ...(catalog !== undefined ? { catalog } : {}), + description, + ...(context !== undefined && context.length > 0 ? { context } : {}), + prompt, + ...(goals.length > 0 ? { goals } : {}), + ...(intent !== undefined ? { intent } : {}), + ...(successCriteria.length > 0 ? { successCriteria } : {}), + ...(doNot.length > 0 ? { doNot } : {}), + ...(reportFocus !== undefined && reportFocus.length > 0 ? { reportFocus } : {}), + signal: childCtl.signal, + onEvent, + ...(deps.onProgress !== undefined ? { onProgress: deps.onProgress } : {}), + ...(resolved.capabilities !== undefined ? { capabilities: resolved.capabilities } : {}), + systemPromptRole: resolved.systemPromptRole, + directorId: resolved.directorId, + maxTurns: resolvedMaxTurns, + }; + + // Fire and forget: this handler must return before the worker finishes. + // The worker's outcome lands in deps.sessions, which wait_agents reads. + deps + .run(params) + .then((result) => { + if (childCtl.signal.aborted) return; + deps.sessions.complete(session.id, result); + }) + .catch((err) => { + if (childCtl.signal.aborted) return; + const message = err instanceof Error ? err.message : String(err); + deps.sessions.fail(session.id, message); + }) + .finally(() => { + telemetry.capture("subagent_end", { + agent_name: agentName, + status: deps.sessions.get(session.id)?.status ?? "completed", + duration_ms: Date.now() - startedAt, + }); + }); + + return fleetResult(call.id, JSON.stringify({ agent_id: session.id, status: "running" })); + }, + }); +} + +interface WaitAgentsDeps { + sessions: SubAgentSessionStore; +} + +function isTerminal(status: SubAgentSessionStatus): boolean { + return status !== "running"; +} + +/** + * Blocks until any of `targets` is terminal, or `timeoutMs` elapses. + * Driven by the session store's mailbox (`subscribe`), raced against a + * timer — never polls. Returns without side effects on timeout: the + * targets are not touched, so a caller can wait again immediately. + */ +async function waitForAnyTerminal( + sessions: SubAgentSessionStore, + targets: readonly string[], + timeoutMs: number, +): Promise { + const alreadyTerminal = targets.some((id) => { + const s = sessions.get(id); + return s !== undefined && isTerminal(s.status); + }); + if (alreadyTerminal) return false; + + return await new Promise((resolve) => { + let settled = false; + const finish = (timedOut: boolean): void => { + if (settled) return; + settled = true; + clearTimeout(timer); + unsubscribe(); + resolve(timedOut); + }; + const timer = setTimeout(() => finish(true), timeoutMs); + const unsubscribe = sessions.subscribe(() => { + if ( + targets.some((id) => { + const s = sessions.get(id); + return s !== undefined && isTerminal(s.status); + }) + ) { + finish(false); + } + }); + }); +} + +export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { + return tool({ + definition: waitAgentsToolDefinition, + 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 targets = + parsed.targets !== undefined && parsed.targets.length > 0 + ? parsed.targets + : deps.sessions + .list() + .filter((s) => s.status === "running") + .map((s) => s.id); + + if (targets.length === 0) { + return fleetResult(call.id, JSON.stringify({ results: [], timed_out: false })); + } + + const timedOut = await waitForAnyTerminal(deps.sessions, targets, timeoutMs); + + const results = targets.map((id) => { + const session = deps.sessions.get(id); + if (session === undefined) { + return { agent_id: id, status: "unknown" as const }; + } + return { + agent_id: id, + status: session.status, + ...(session.report !== undefined ? { report: session.report } : {}), + ...(session.error !== undefined ? { error: session.error } : {}), + }; + }); + + return fleetResult(call.id, JSON.stringify({ results, timed_out: timedOut })); + }, + }); +} diff --git a/src/subagent/run.ts b/src/subagent/run.ts index 942070ad..e985384f 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -118,6 +118,8 @@ import { isSubAgentCancelError, } from "./dispose.js"; import { createTaskTool } from "./task-tool.js"; +import { createSpawnAgentTool, createWaitAgentsTool } from "./agent-fleet.js"; +import { createSubAgentSessionStore } from "./session-store.js"; import type { RunSubAgentParams, SubAgentProvider } from "./types.js"; import type { TaskIntent } from "./report.js"; import { runWithSubAgentIdentity } from "./identity-context.js"; @@ -486,7 +488,13 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { // AgentProfile with orchestrator: true from mounting task/search_agents // just because it is outside the closed director set. const tier = params.orchestratorTier ?? "leaf"; - for (const verb of ["task", "search_agents", "read_agent_trace"]) { + for (const verb of [ + "task", + "search_agents", + "read_agent_trace", + "spawn_agent", + "wait_agents", + ]) { assertTierMayMountFleetVerb(tier, verb); } if (params.nestedDispatch === undefined) { @@ -546,6 +554,33 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { getNodes: () => nd.sessions?.list() ?? [], }), ]; + // spawn_agent/wait_agents (CL-6942) need a session store as their + // mailbox; reuse the orchestrator's if it has one, else give this + // install its own rather than inventing separate bookkeeping. + const fleetSessions = nd.sessions ?? createSubAgentSessionStore(); + const fleetDeps = { + permissionGate: nd.permissionGate, + ...(nd.inheritMcpTools !== undefined ? { inheritMcpTools: nd.inheritMcpTools } : {}), + ...(nd.shellTimeout !== undefined ? { shellTimeout: nd.shellTimeout } : {}), + ...(nd.shellEnv !== undefined ? { shellEnv: nd.shellEnv } : {}), + ...(nd.extraToolPlugins !== undefined ? { extraToolPlugins: nd.extraToolPlugins } : {}), + cwd: params.cwd, + getWorkdirBase: nd.getWorkdirBase, + provider: nd.provider, + getBlobReader: () => sessionBlobReader, + run: runSubAgent, + telemetry: liveTelemetry, + sessions: fleetSessions, + ...(nd.onEvent !== undefined ? { onEvent: nd.onEvent } : {}), + ...(nd.onProgress !== undefined ? { onProgress: nd.onProgress } : {}), + ...(nd.settings !== undefined ? { settings: nd.settings } : {}), + ...(nd.catalog !== undefined ? { catalog: nd.catalog } : {}), + }; + tools = [ + ...tools, + createSpawnAgentTool(fleetDeps), + createWaitAgentsTool({ sessions: fleetSessions }), + ]; } const environment = await gatherEnvironment(params.cwd); From d8d4dfbe003ff41647a8f601558982bd3c8426c6 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 23 Aug 2026 22:34:16 -0700 Subject: [PATCH 2/4] Fix report eviction and add cwd write-lane refusal for spawn_agent fleetRecords is a never-capped map of terminal spawn_agent results, written before the session store's complete()/fail() so wait_agents never loses a report to the store's TUI-sized finished-session cap. spawn_agent also now refuses a second concurrent implement-intent spawn against the same cwd (no worktree isolation yet), releasing the lane once the running one finishes; explore/plan/review-intent spawns are unaffected and may still run concurrently. --- src/subagent/agent-fleet.test.ts | 134 ++++++++++++++++++++-- src/subagent/agent-fleet.ts | 188 +++++++++++++++++++++++++------ src/subagent/run.ts | 10 +- 3 files changed, 286 insertions(+), 46 deletions(-) diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index ac43c517..641eeb19 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -1,6 +1,11 @@ import { describe, expect, test } from "bun:test"; -import { createSpawnAgentTool, createWaitAgentsTool, type AgentFleetDeps } from "./agent-fleet.js"; +import { + createFleetRecords, + createSpawnAgentTool, + createWaitAgentsTool, + type AgentFleetDeps, +} from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; import { createPermissionGate } from "../permission/gate.js"; import type { RunSubAgentParams } from "./types.js"; @@ -31,21 +36,25 @@ function deferred(): { return { promise, resolve, reject }; } -function makeDeps(run: (params: RunSubAgentParams) => Promise): AgentFleetDeps { +function makeDeps( + run: (params: RunSubAgentParams) => Promise, + opts: { cwd?: string } = {}, +): AgentFleetDeps { return { permissionGate: testPermissionGate, - cwd: "/tmp", + cwd: opts.cwd ?? "/tmp", getWorkdirBase: () => "/tmp/workdir", provider, run, sessions: createSubAgentSessionStore(), + fleetRecords: createFleetRecords(), }; } -async function callTool( +async function callToolRaw( tool: ReturnType | ReturnType, args: Record, -): Promise> { +): Promise<{ content: string; isError?: boolean }> { if (tool.kind !== "full") throw new Error(`expected full tool, got ${tool.kind}`); const result = await tool.handler( { id: `call-${Math.random()}`, name: tool.definition.name, arguments: args }, @@ -53,6 +62,14 @@ async function callTool( ); const content = typeof result.content === "string" ? result.content : JSON.stringify(result.content); + return { content, ...(result.isError !== undefined ? { isError: result.isError } : {}) }; +} + +async function callTool( + tool: ReturnType | ReturnType, + args: Record, +): Promise> { + const { content } = await callToolRaw(tool, args); return JSON.parse(content); } @@ -90,7 +107,7 @@ describe("spawn_agent + wait_agents", () => { return gates[i]!.promise; }); const spawn = createSpawnAgentTool(deps); - const wait = createWaitAgentsTool({ sessions: deps.sessions }); + const wait = createWaitAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords }); const spawned = await Promise.all( [0, 1, 2].map((i) => @@ -120,7 +137,7 @@ describe("spawn_agent + wait_agents", () => { const gate = deferred(); const deps = makeDeps(async () => gate.promise); const spawn = createSpawnAgentTool(deps); - const wait = createWaitAgentsTool({ sessions: deps.sessions }); + const wait = createWaitAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords }); const spawned = await callTool(spawn, { description: "slow job", @@ -155,7 +172,7 @@ describe("spawn_agent + wait_agents", () => { let callIndex = 0; const deps = makeDeps(async () => gates[callIndex++]!.promise); const spawn = createSpawnAgentTool(deps); - const wait = createWaitAgentsTool({ sessions: deps.sessions }); + const wait = createWaitAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords }); await callTool(spawn, { description: "a", prompt: "do it", intent: "explore" }); await callTool(spawn, { description: "b", prompt: "do it", intent: "explore" }); @@ -169,4 +186,105 @@ describe("spawn_agent + wait_agents", () => { gates[1]!.resolve("b done"); }); + + test("reports survive well past the session store's display cap (20) until wait_agents collects them", async () => { + // DEFAULT_MAX_COMPLETED on SubAgentSessionStore is 20 finished sessions; + // spawn (and complete) enough workers to blow well past it before any of + // them is collected, proving fleetRecords — not the store — is what + // wait_agents actually reads from. + const COUNT = 25; + const deps = makeDeps(async () => "irrelevant"); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords }); + + const ids: string[] = []; + for (let i = 0; i < COUNT; i++) { + const spawned = await callTool(spawn, { + description: `job-${i}`, + prompt: `report-${i}`, + intent: "explore", + }); + ids.push(spawned.agent_id as string); + } + + // Let every spawn's run() resolve and complete() land before collecting. + await new Promise((resolve) => setTimeout(resolve, 20)); + + // The store itself has already evicted all but the most recent 20. + expect(deps.sessions.get(ids[0]!)).toBeUndefined(); + + // But every single one is still retrievable through wait_agents. + const waited = await callTool(wait, { targets: ids, timeout_ms: 5000 }); + const results = waited.results as { agent_id: string; status: string; report?: string }[]; + expect(results).toHaveLength(COUNT); + for (const result of results) { + expect(result.status).toBe("done"); + expect(result.report).toBe("irrelevant"); + } + }); +}); + +describe("spawn_agent write-lane isolation", () => { + test("refuses a second concurrent implement-intent spawn against the same cwd", async () => { + const gate = deferred(); + const deps = makeDeps(async () => gate.promise, { cwd: "/repo" }); + const spawn = createSpawnAgentTool(deps); + + const first = await callTool(spawn, { + description: "build one", + prompt: "implement thing one", + intent: "implement", + }); + expect(first.status).toBe("running"); + + const second = await callToolRaw(spawn, { + description: "build two", + prompt: "implement thing two", + intent: "implement", + }); + expect(second.isError).toBe(true); + expect(second.content).toContain("Error:"); + expect(second.content).toContain(first.agent_id as string); + + gate.resolve("done"); + }); + + test("does not refuse a second concurrent explore-intent spawn against the same cwd", async () => { + const deps = makeDeps(async () => "explored", { cwd: "/repo" }); + const spawn = createSpawnAgentTool(deps); + + const first = await callTool(spawn, { + description: "explore one", + prompt: "look around", + intent: "explore", + }); + const second = await callTool(spawn, { + description: "explore two", + prompt: "look around more", + intent: "explore", + }); + + expect(first.status).toBe("running"); + expect(second.status).toBe("running"); + }); + + test("releases the write lane once the implement worker finishes, allowing another", async () => { + const deps = makeDeps(async () => "built", { cwd: "/repo" }); + const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords }); + + const first = await callTool(spawn, { + description: "build one", + prompt: "implement thing one", + intent: "implement", + }); + await callTool(wait, { targets: [first.agent_id as string], timeout_ms: 5000 }); + + const second = await callTool(spawn, { + description: "build two", + prompt: "implement thing two", + intent: "implement", + }); + expect(second.status).toBe("running"); + }); }); diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 5e448200..70610af7 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -8,18 +8,36 @@ * ones it cares about (wait_agents), instead of one task() call per worker * serializing the wait. * - * State lives entirely in the existing SubAgentSessionStore (status/report/ - * error, keyed by session id) — no parallel bookkeeping map. wait_agents' - * blocking is driven by the store's `subscribe` mailbox raced against a - * timeout timer; it never polls. + * Running state and the mailbox (`subscribe`) are the existing + * SubAgentSessionStore's — wait_agents' blocking is driven by that + * `subscribe` raced against a timeout timer, never polling. But the store's + * finished-session retention is a TUI display cap (`maxCompleted`, default + * 20): `complete()`/`fail()` evict the oldest finished session — report and + * all — once more than that many have finished. task() never hit this + * because it awaits its own single result before the tool call returns; here + * a caller can spawn far more workers than the cap in one turn and only + * `wait_agents` them later, so an evicted report would otherwise vanish + * silently. `fleetRecords` below is a small, deliberately-separate map + * (agent id -> terminal status/report/error) that is never capped and is + * only ever cleared when `wait_agents` actually delivers that result to a + * caller — it exists precisely because the store's cap cannot be trusted for + * this use. * * Argument shape intentionally mirrors `task()`'s (description/prompt/ * context/goals/intent/success_criteria/do_not/report_focus/maxTurns) so a * caller can swap one for the other. Scope is deliberately narrower than * `task()` for this first cut: only closed-director dispatch (`agent=` a * director id, or `intent=`) is supported — no custom AgentProfile lookup, - * no worktree isolation, no nested orchestration, no re-dispatch ledger. - * Those are `task()`-only for now; nothing here stops adding them later. + * no nested orchestration, no re-dispatch ledger. Those remain `task()`-only + * for now; nothing here stops adding them later. + * + * Worktree isolation: task() supports it, spawn_agent does not (yet). Since + * spawn_agent's whole point is running several workers at once, two workers + * sharing one cwd with write intent would silently corrupt each other's + * edits. Rather than duplicate task()'s worktree machinery here, spawn_agent + * refuses a second concurrent implement-intent (director "build") spawn + * against the same cwd with an actionable error — explore/plan/review + * workers, which do not write, are unaffected and may run concurrently. */ import { tool } from "@intx/agent"; @@ -43,11 +61,61 @@ import { resolveSubAgentMaxTurns, validateTaskMaxTurns } from "../config/setting import { resolveEffortForRole } from "../provider/reasoning-effort.js"; import { isCodexProviderName } from "../config/codex-providers.js"; import { buildDispatchBrief, type TaskIntent } from "./report.js"; -import type { SubAgentSessionStore, SubAgentSessionStatus } from "./session-store.js"; +import type { SubAgentSessionStore } from "./session-store.js"; import type { RunSubAgentParams, SubAgentProvider, SubAgentSandboxDeps } from "./types.js"; import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js"; import { classifyAgentName } from "../telemetry/classify.js"; +/** Terminal (or running) record for one spawned agent, keyed by agent id. */ +interface FleetRecord { + status: "running" | "done" | "failed"; + report?: string; + error?: string; +} + +/** + * Never-capped terminal-result store, cleared only once a result is + * delivered to a wait_agents caller. See the module doc comment for why the + * session store's own retention cannot be reused here. + */ +class FleetRecords { + private readonly records = new Map(); + + register(id: string): void { + this.records.set(id, { status: "running" }); + } + + resolve(id: string, report: string): void { + this.records.set(id, { status: "done", report }); + } + + reject(id: string, error: string): void { + this.records.set(id, { status: "failed", error }); + } + + /** Read without consuming — used for the terminal-yet check. */ + peek(id: string): FleetRecord | undefined { + return this.records.get(id); + } + + /** Read and, if terminal, remove — a delivered result is not kept around. */ + take(id: string): FleetRecord | undefined { + const record = this.records.get(id); + if (record !== undefined && record.status !== "running") { + this.records.delete(id); + } + return record; + } +} + +// One registry per orchestrator install (shared by its spawn_agent and +// wait_agents tool instances), not a module singleton — created in +// createSpawnAgentTool and threaded to createWaitAgentsTool by the caller. +export type FleetRecordsHandle = FleetRecords; +export function createFleetRecords(): FleetRecordsHandle { + return new FleetRecords(); +} + const SpawnAgentArgs = type({ description: "string", prompt: "string", @@ -146,6 +214,7 @@ export type AgentFleetDeps = SubAgentSandboxDeps & { provider: SubAgentProvider | (() => SubAgentProvider); run: (params: RunSubAgentParams) => Promise; sessions: SubAgentSessionStore; + fleetRecords: FleetRecordsHandle; settings?: Settings | (() => Settings | undefined); catalog?: readonly ProviderCatalogEntry[] | (() => readonly ProviderCatalogEntry[]); onEvent?: (event: ReactorEmittedEvent) => void; @@ -212,8 +281,22 @@ function resolveDirectorDispatch( }; } +/** + * Director ids that write. Only "build" (the implement-intent director) + * needs cwd exclusivity today; explore/plan/review/critique-style directors + * do not write and may run concurrently against the same cwd. + */ +function isWriteRiskDirector(directorId: string): boolean { + return directorId === "build"; +} + export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { const telemetry = deps.telemetry ?? NOOP_TELEMETRY; + // cwd -> agent ids of running write-risk (implement) workers against it. + // deps.cwd is fixed for the lifetime of this tool instance (one per + // orchestrator install), so this only ever guards concurrent spawns from + // the same orchestrator turn, which is exactly the case with no isolation. + const writeLanes = new Map>(); return tool({ definition: spawnAgentToolDefinition, handler: async (call, _signal): Promise => { @@ -253,6 +336,20 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { const resolved = resolveDirectorDispatch(agentId, intent); if (!resolved.ok) return fleetResult(call.id, resolved.error); + const isWriteRisk = isWriteRiskDirector(resolved.directorId); + if (isWriteRisk) { + const lane = writeLanes.get(deps.cwd); + if (lane !== undefined && lane.size > 0) { + return fleetResult( + call.id, + `Error: spawn_agent refused — an implement-intent worker (${[...lane].join(", ")}) is ` + + `already running against ${deps.cwd} and spawn_agent has no worktree isolation yet, so a ` + + `second one would risk corrupting the first one's edits. Wait for it via wait_agents first, ` + + `or use task(useWorktree: true) for isolated concurrent implementation work.`, + ); + } + } + let taskMaxTurns: number | undefined; if (rawMaxTurns !== undefined) { const verdict = validateTaskMaxTurns(rawMaxTurns); @@ -293,6 +390,15 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { agentId: resolved.directorId, brief, }); + deps.fleetRecords.register(session.id); + if (isWriteRisk) { + let lane = writeLanes.get(deps.cwd); + if (lane === undefined) { + lane = new Set(); + writeLanes.set(deps.cwd, lane); + } + lane.add(session.id); + } const agentName = classifyAgentName(resolved.directorId); telemetry.capture("subagent_start", { agent_name: agentName }); const startedAt = Date.now(); @@ -338,16 +444,28 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }; // Fire and forget: this handler must return before the worker finishes. - // The worker's outcome lands in deps.sessions, which wait_agents reads. + // fleetRecords (never capped) is the durable source of truth wait_agents + // reads from; deps.sessions.complete/fail is still called for the TUI's + // benefit, but only after fleetRecords already has the result, and + // fleetRecords is written before it so the synchronous subscribe + // notification fired by complete()/fail() always sees the up-to-date + // record. + const releaseWriteLane = (): void => { + if (isWriteRisk) writeLanes.get(deps.cwd)?.delete(session.id); + }; deps .run(params) .then((result) => { + releaseWriteLane(); if (childCtl.signal.aborted) return; + deps.fleetRecords.resolve(session.id, result); deps.sessions.complete(session.id, result); }) .catch((err) => { + releaseWriteLane(); if (childCtl.signal.aborted) return; const message = err instanceof Error ? err.message : String(err); + deps.fleetRecords.reject(session.id, message); deps.sessions.fail(session.id, message); }) .finally(() => { @@ -365,28 +483,29 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { interface WaitAgentsDeps { sessions: SubAgentSessionStore; -} - -function isTerminal(status: SubAgentSessionStatus): boolean { - return status !== "running"; + fleetRecords: FleetRecordsHandle; } /** - * Blocks until any of `targets` is terminal, or `timeoutMs` elapses. - * Driven by the session store's mailbox (`subscribe`), raced against a - * timer — never polls. Returns without side effects on timeout: the - * targets are not touched, so a caller can wait again immediately. + * 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. */ async function waitForAnyTerminal( sessions: SubAgentSessionStore, + fleetRecords: FleetRecordsHandle, targets: readonly string[], timeoutMs: number, ): Promise { - const alreadyTerminal = targets.some((id) => { - const s = sessions.get(id); - return s !== undefined && isTerminal(s.status); - }); - if (alreadyTerminal) return false; + const anyTerminal = (): boolean => + targets.some((id) => { + const record = fleetRecords.peek(id); + return record !== undefined && record.status !== "running"; + }); + if (anyTerminal()) return false; return await new Promise((resolve) => { let settled = false; @@ -399,14 +518,7 @@ async function waitForAnyTerminal( }; const timer = setTimeout(() => finish(true), timeoutMs); const unsubscribe = sessions.subscribe(() => { - if ( - targets.some((id) => { - const s = sessions.get(id); - return s !== undefined && isTerminal(s.status); - }) - ) { - finish(false); - } + if (anyTerminal()) finish(false); }); }); } @@ -434,18 +546,24 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { return fleetResult(call.id, JSON.stringify({ results: [], timed_out: false })); } - const timedOut = await waitForAnyTerminal(deps.sessions, targets, timeoutMs); + const timedOut = await waitForAnyTerminal(deps.sessions, deps.fleetRecords, targets, timeoutMs); + // Terminal records are consumed (removed) once delivered here; a + // running record is only peeked, so it stays waitable. const results = targets.map((id) => { - const session = deps.sessions.get(id); - if (session === undefined) { + const record = deps.fleetRecords.peek(id); + 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: session.status, - ...(session.report !== undefined ? { report: session.report } : {}), - ...(session.error !== undefined ? { error: session.error } : {}), + status: taken.status, + ...(taken.report !== undefined ? { report: taken.report } : {}), + ...(taken.error !== undefined ? { error: taken.error } : {}), }; }); diff --git a/src/subagent/run.ts b/src/subagent/run.ts index e985384f..3f11bb9b 100644 --- a/src/subagent/run.ts +++ b/src/subagent/run.ts @@ -118,7 +118,7 @@ import { isSubAgentCancelError, } from "./dispose.js"; import { createTaskTool } from "./task-tool.js"; -import { createSpawnAgentTool, createWaitAgentsTool } from "./agent-fleet.js"; +import { createFleetRecords, createSpawnAgentTool, createWaitAgentsTool } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; import type { RunSubAgentParams, SubAgentProvider } from "./types.js"; import type { TaskIntent } from "./report.js"; @@ -556,8 +556,11 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { ]; // spawn_agent/wait_agents (CL-6942) need a session store as their // mailbox; reuse the orchestrator's if it has one, else give this - // install its own rather than inventing separate bookkeeping. + // install its own. fleetRecords is a small never-capped map for + // terminal results the session store's display cap would otherwise + // evict before wait_agents collects them (see agent-fleet.ts). const fleetSessions = nd.sessions ?? createSubAgentSessionStore(); + const fleetRecords = createFleetRecords(); const fleetDeps = { permissionGate: nd.permissionGate, ...(nd.inheritMcpTools !== undefined ? { inheritMcpTools: nd.inheritMcpTools } : {}), @@ -571,6 +574,7 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { run: runSubAgent, telemetry: liveTelemetry, sessions: fleetSessions, + fleetRecords, ...(nd.onEvent !== undefined ? { onEvent: nd.onEvent } : {}), ...(nd.onProgress !== undefined ? { onProgress: nd.onProgress } : {}), ...(nd.settings !== undefined ? { settings: nd.settings } : {}), @@ -579,7 +583,7 @@ export async function runSubAgent(params: RunSubAgentParams): Promise { tools = [ ...tools, createSpawnAgentTool(fleetDeps), - createWaitAgentsTool({ sessions: fleetSessions }), + createWaitAgentsTool({ sessions: fleetSessions, fleetRecords }), ]; } From a69fe2a3a7e0f23fba9de2af0acf4e4883e782e3 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 23 Aug 2026 22:36:14 -0700 Subject: [PATCH 3/4] Format agent-fleet.ts --- src/subagent/agent-fleet.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 70610af7..8eb6685e 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -546,7 +546,12 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { return fleetResult(call.id, JSON.stringify({ results: [], timed_out: false })); } - const timedOut = await waitForAnyTerminal(deps.sessions, deps.fleetRecords, targets, timeoutMs); + const timedOut = await waitForAnyTerminal( + deps.sessions, + deps.fleetRecords, + targets, + timeoutMs, + ); // Terminal records are consumed (removed) once delivered here; a // running record is only peeked, so it stays waitable. From 7b229eb6578b3cf7f585b408cd2a40f3d7955b06 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 23 Aug 2026 22:48:34 -0700 Subject: [PATCH 4/4] Name spawned workers' trace dirs after their session id read_agent_trace's descendant-scoping check resolves a worker's parent chain from its trace directory name, so spawn_agent must pass the session-store id the same way task-tool.ts does. --- src/subagent/agent-fleet.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 8eb6685e..3c64cafc 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -415,6 +415,10 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { const catalog = deps.catalog !== undefined ? resolveDep(deps.catalog) : undefined; const params: RunSubAgentParams = { + // Name the trace directory after the session-store id so the + // descendant-scoping check behind read_agent_trace can resolve this + // worker's parent chain (matches task-tool.ts). + id: session.id, permissionGate: deps.permissionGate, ...(deps.inheritMcpTools !== undefined ? { inheritMcpTools: deps.inheritMcpTools } : {}), ...(deps.shellTimeout !== undefined ? { shellTimeout: deps.shellTimeout } : {}),