Skip to content

Remove same-cwd spawn refusal; bound fleetRecords memory #617

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheGreatAxios merged 1 commit into from
Aug 24, 2026
Merged

Remove same-cwd spawn refusal; bound fleetRecords memory #617

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
101 changes: 62 additions & 39 deletions src/subagent/agent-fleet.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import {
createFleetRecords,
createSpawnAgentTool,
createWaitAgentsTool,
MAX_FLEET_RECORDS,
type AgentFleetDeps,
} from "./agent-fleet.js";
import { createSubAgentSessionStore } from "./session-store.js";
Expand Down Expand Up @@ -228,67 +229,89 @@ 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<RunSubAgentResult>();
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<RunSubAgentResult>(), deferred<RunSubAgentResult>()];
let callIndex = 0;
const deps = makeDeps(async () => gates[callIndex++]!.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, {
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");
});
});
118 changes: 62 additions & 56 deletions src/subagent/agent-fleet.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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";
Expand Down Expand Up @@ -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<string, FleetRecord>();
Expand All @@ -92,25 +104,59 @@ 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. */
peek(id: string): FleetRecord | undefined {
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
Expand Down Expand Up @@ -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<string, Set<string>>();
return tool({
definition: spawnAgentToolDefinition,
handler: async (call, _signal): Promise<ToolResult> => {
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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 } : {}),
};
});

Expand Down
Loading