diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b7c92f6b..1998a9b42 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,15 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename omitted-text branch that unconditionally completed a tool-less turn is removed, so every call path gets the `incomplete-report` nudge and salvage when a tool-using run ends in envelope-less narration. +- `spawn_agent` no longer refuses a second concurrent implement-intent spawn + against the same working directory — running multiple agents against one + worktree is allowed by design, and the refusal was guarding against churn + that resolves on its own, not corruption. +- `fleetRecords` (the store behind `wait_agents`) now caps how many full + reports it holds in memory; past the cap, the oldest report already + delivered to a caller is compacted to a status-only tombstone pointing at + `read_agent_trace` for the detail, so an uncollected report is never + evicted ahead of one that's already been picked up. ## [0.2.109] - 2026-08-24 diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index 17e4754eb..de59d1d12 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -4,6 +4,7 @@ import { createFleetRecords, createSpawnAgentTool, createWaitAgentsTool, + MAX_FLEET_RECORDS, type AgentFleetDeps, } from "./agent-fleet.js"; import { createSubAgentSessionStore } from "./session-store.js"; @@ -228,10 +229,11 @@ describe("spawn_agent + wait_agents", () => { }); }); -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" }); +describe("spawn_agent same-cwd concurrency", () => { + test("two concurrent implement-intent spawn_agent calls against the same cwd both start", async () => { + const gates = [deferred(), deferred()]; + let callIndex = 0; + const deps = makeDeps(async () => gates[callIndex++]!.promise, { cwd: "/repo" }); const spawn = createSpawnAgentTool(deps); const first = await callTool(spawn, { @@ -239,56 +241,77 @@ describe("spawn_agent write-lane isolation", () => { prompt: "implement thing one", intent: "implement", }); - expect(first.status).toBe("running"); - - const second = await callToolRaw(spawn, { + const second = await callTool(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({ report: "done" }); + expect(first.status).toBe("running"); + expect(second.status).toBe("running"); + + gates[0]!.resolve({ report: "one done" }); + gates[1]!.resolve({ report: "two done" }); }); +}); - test("does not refuse a second concurrent explore-intent spawn against the same cwd", async () => { - const deps = makeDeps(async () => ({ report: "explored" }), { cwd: "/repo" }); +describe("fleetRecords retention cap", () => { + test("many spawned-and-completed workers whose reports are never collected leave memory bounded", async () => { + const COUNT = MAX_FLEET_RECORDS + 50; + const deps = makeDeps(async () => ({ report: "x".repeat(1000) })); const spawn = createSpawnAgentTool(deps); + const wait = createWaitAgentsTool({ sessions: deps.sessions, fleetRecords: deps.fleetRecords }); - 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", - }); + const ids: string[] = []; + for (let i = 0; i < COUNT; i++) { + const spawned = await callTool(spawn, { + description: `job-${i}`, + prompt: `p-${i}`, + intent: "explore", + }); + ids.push(spawned.agent_id as string); + } + await new Promise((resolve) => setTimeout(resolve, 20)); - expect(first.status).toBe("running"); - expect(second.status).toBe("running"); + const waited = await callTool(wait, { targets: ids, timeout_ms: 5000 }); + const results = waited.results as { status: string; report?: string }[]; + const withReport = results.filter((r) => r.report !== undefined).length; + + // Payloads are capped: well under COUNT full reports survive uncollected. + expect(withReport).toBeLessThanOrEqual(MAX_FLEET_RECORDS); + expect(withReport).toBeLessThan(COUNT); }); - test("releases the write lane once the implement worker finishes, allowing another", async () => { - const deps = makeDeps(async () => ({ report: "built" }), { cwd: "/repo" }); + test("an evicted-but-uncollected agent resolves to its terminal status plus a read_agent_trace pointer", async () => { + const COUNT = MAX_FLEET_RECORDS + 50; + const deps = makeDeps(async () => ({ report: "x".repeat(1000) })); 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 ids: string[] = []; + for (let i = 0; i < COUNT; i++) { + const spawned = await callTool(spawn, { + description: `job-${i}`, + prompt: `p-${i}`, + intent: "explore", + }); + ids.push(spawned.agent_id as string); + } + await new Promise((resolve) => setTimeout(resolve, 20)); - const second = await callTool(spawn, { - description: "build two", - prompt: "implement thing two", - intent: "implement", - }); - expect(second.status).toBe("running"); + // The earliest spawned agent's payload should have been tombstoned — + // never collected, so it was evicted once the cap was exceeded. + const waited = await callTool(wait, { targets: [ids[0]!], timeout_ms: 5000 }); + const results = waited.results as { + agent_id: string; + status: string; + report?: string; + hint?: string; + }[]; + expect(results).toHaveLength(1); + expect(results[0]!.status).not.toBe("unknown"); + expect(["done", "failed"]).toContain(results[0]!.status); + expect(results[0]!.report).toBeUndefined(); + expect(results[0]!.hint).toContain("read_agent_trace"); }); }); diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index eb329522d..875d3f5f4 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -18,10 +18,16 @@ * 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 + * (agent id -> terminal status/report/error), kept alive across the store's + * own eviction and cleared only when `wait_agents` delivers a result to a * caller — it exists precisely because the store's cap cannot be trusted for - * this use. + * this use. Its heavy payloads (report/error text) are capped at + * `MAX_FLEET_RECORDS`: past that, the oldest already-collected entry is + * compacted to a tombstone (status only, plus a pointer at + * `read_agent_trace` for the detail), falling back to the oldest + * uncollected one only once every collected entry is gone — a caller who + * never called wait_agents still gets a terminal status, never a bare + * "unknown". * * Argument shape intentionally mirrors `task()`'s (description/prompt/ * context/goals/intent/success_criteria/do_not/report_focus/maxTurns) so a @@ -31,13 +37,6 @@ * 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"; @@ -76,12 +75,25 @@ interface FleetRecord { status: "running" | "done" | "failed"; report?: string; error?: string; + /** Set once a wait_agents caller has been handed this result. */ + collected?: boolean; + /** Set once the payload has been compacted away to bound memory. */ + tombstoned?: boolean; + /** Present only on a tombstoned record — how to recover the detail. */ + hint?: string; } +const RECOVERY_HINT = + "Report evicted to bound fleet memory; recover full detail via read_agent_trace(agent_id)."; + +/** Payload cap: terminal records still holding a report/error. */ +export const MAX_FLEET_RECORDS = 200; + /** - * 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. + * Terminal-result store, cleared 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, and for the tombstone + * eviction policy once more than `MAX_FLEET_RECORDS` payloads are held. */ class FleetRecords { private readonly records = new Map(); @@ -92,10 +104,12 @@ class FleetRecords { resolve(id: string, report: string): void { this.records.set(id, { status: "done", report }); + this.enforceCap(); } reject(id: string, error: string): void { this.records.set(id, { status: "failed", error }); + this.enforceCap(); } /** Read without consuming — used for the terminal-yet check. */ @@ -103,14 +117,46 @@ class FleetRecords { return this.records.get(id); } - /** Read and, if terminal, remove — a delivered result is not kept around. */ + /** + * Read and, if terminal, mark collected. The entry is kept (not deleted) + * so a later query still resolves to a real status instead of "unknown" — + * it just becomes the preferred eviction target once the payload cap is + * hit. + */ take(id: string): FleetRecord | undefined { const record = this.records.get(id); if (record !== undefined && record.status !== "running") { - this.records.delete(id); + record.collected = true; } return record; } + + private hasPayload(record: FleetRecord): boolean { + return record.status !== "running" && !record.tombstoned; + } + + /** + * Compacts the oldest already-collected payload to a tombstone first — + * its caller already has the detail — and only reaches into uncollected + * payloads once no collected one remains. + */ + private enforceCap(): void { + let payloadCount = 0; + for (const record of this.records.values()) { + if (this.hasPayload(record)) payloadCount++; + } + while (payloadCount > MAX_FLEET_RECORDS) { + const victim = + [...this.records.values()].find((r) => this.hasPayload(r) && r.collected === true) ?? + [...this.records.values()].find((r) => this.hasPayload(r)); + if (victim === undefined) break; + delete victim.report; + delete victim.error; + victim.tombstoned = true; + victim.hint = RECOVERY_HINT; + payloadCount--; + } + } } // One registry per orchestrator install (shared by its spawn_agent and @@ -286,22 +332,8 @@ 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 => { @@ -341,20 +373,6 @@ 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); @@ -396,14 +414,6 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { 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(); @@ -459,19 +469,14 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { // 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.report); deps.sessions.complete(session.id, result.report); }) .catch((err) => { - releaseWriteLane(); if (childCtl.signal.aborted) return; const message = err instanceof Error ? err.message : String(err); deps.fleetRecords.reject(session.id, message); @@ -578,6 +583,7 @@ export function createWaitAgentsTool(deps: WaitAgentsDeps): AgentTool { status: taken.status, ...(taken.report !== undefined ? { report: taken.report } : {}), ...(taken.error !== undefined ? { error: taken.error } : {}), + ...(taken.hint !== undefined ? { hint: taken.hint } : {}), }; });