diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index 55a90096c..c295f0cee 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -167,7 +167,7 @@ Sixteen packages under `src/agent/directors//` register in `DIRECTOR_REGISTR **Codex tool proxies.** When the active provider is Codex (`isCodexProviderName`), `createAgentToolset` and `runSubAgent` mount `apply_patch`, `shell`, and `update_plan` stringTools from `createCodexToolProxies`, all forwarding through the same posix `ToolRunner` seam (`runTool`) so permission plugins still apply. `apply_patch` parses the Codex envelope and forwards each op (`write_file` / `delete_file` / `read_file`). `shell` — the native Codex name is `shell`, not `exec_command` — normalizes Codex's `command` (string or `["bash","-lc",script]`-style argv array), `workdir`, and `timeout_ms` onto `run_shell`'s `{command, cwd?, timeout?}` and is gated by `allowShellFromCapabilities` (mirrors `allowDeleteFromCapabilities` against `run_shell`). `update_plan` maps Codex's `plan: [{step, status}]` onto `manage_tasks(action: "create")`; `pending`/`in_progress`/`completed` map to `todo`/`doing`/`done` — `manage_tasks`'s `cancelled` status has no Codex equivalent and is never produced by this proxy. Primary strips `apply_patch` after mount (Corbits DIY stays on `write_file` / `edit_file` / `delete_file`); `shell` and `update_plan` stay on primary (same classification as `run_shell` / `manage_tasks`). Build and docs worker allowlists (`BUILD_TOOLS` / `DOCS_TOOLS`) include `apply_patch` so Codex workers keep the proxy after the capability filter. `CORE_TOOL_NAMES` does not list it. -6. There is no static write-path declaration on packages or profiles (CL-6952 removed it — no shipped director ever set one). Instead, `agent-fleet.ts` tracks each running dispatch by cwd; a new dispatch that lands on the same cwd as a still-running lane records a `concurrent-lane-overlap` entry in `intervention-log.ts` (class `conflict`). This is advisory only — it never blocks the spawn, since cwd overlap does not prove the two lanes touch the same files. +6. There is no static write-path declaration on packages or profiles (CL-6952 removed it — no shipped director ever set one). Instead, `agent-fleet.ts` tracks each running dispatch by cwd; a new mutating dispatch that lands on the same cwd as a live mutating peer (`pending_init`/`running`, and not a declared read-only `modelRole` of `explore`/`plan`/`review`/`test`) records at most one `concurrent-lane-overlap` entry per cwd wave in `intervention-log.ts` (class `conflict`). The wave flag clears when no live mutating writer remains for that cwd. Terminal-but-unsettled lanes (for example cancelled with `finishedAt` set while the run promise has not reached `finally`) are pruned from the map and do not warn. This is advisory only — it never blocks the spawn, since cwd overlap does not prove the two lanes touch the same files. 7. Spawn effort: pin > package `modelRole` default (`defaultEffortForDirector`; intern=low; plan/review/orchestrator=high; implement/explore/docs/test=medium) > orchestrator/worker binary > parent inheritance. Optional skills are listed in the identity header for awareness; workers do not mount `use_skill` (guidance is baked into package system prompts). Primary mounts `use_skill` for its own skill list. Intent defaults: `intent=implement` → director `builder`; `explore` → `explorer`; `plan` → `counsel`; `review` → `critic`; general → error. Spawn: skywalker full fleet; greybeard intern/explorer/critic only; all other directors mount no fleet tools. Live `` injects cwd, platform, arch, runtime, date, and git status on every chat and worker prompt. diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index a5d96f0f0..75cbaba19 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -664,6 +664,45 @@ describe("spawn_agent same-cwd concurrency", () => { defined(gates[1]).resolve({ report: "two done" }); }); + test("a terminal but unsettled shared-cwd lane does not conflict with a later spawn", async () => { + const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-terminal-")); + const gates = [ + deferred(), + deferred(), + ]; + let callIndex = 0; + const deps = makeDeps(async () => defined(gates[callIndex++]).promise, { + cwd: "/repo", + }); + deps.getWorkdirBase = () => dir; + const spawn = createSpawnAgentTool(deps); + + const first = await callTool(spawn, { + description: "build one", + prompt: "implement thing one", + intent: "implement", + success_criteria: ["thing one ships"], + }); + const firstId = first.agent_id as string; + expect(deps.sessions.cancel(firstId)).toBe(true); + expect(deps.sessions.get(firstId)?.finishedAt).toBeNumber(); + + await callTool(spawn, { + description: "build two", + prompt: "implement thing two", + intent: "implement", + success_criteria: ["thing two ships"], + }); + await new Promise((resolve) => setTimeout(resolve, 50)); + + await expect( + readFile(join(dir, INTERVENTION_FILE), "utf8"), + ).rejects.toThrow(); + + defined(gates[0]).resolve({ report: "one cancelled" }); + defined(gates[1]).resolve({ report: "two done" }); + }); + test("two concurrent shared-cwd spawn_agent lanes log concurrent-lane-overlap", async () => { const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-")); const gates = [ @@ -706,10 +745,224 @@ describe("spawn_agent same-cwd concurrency", () => { expect(log).toContain("/repo"); expect(log).toContain("build one"); expect(log).toContain("build two"); + expect( + log + .trim() + .split("\n") + .filter((line) => line.includes("concurrent-lane-overlap")), + ).toHaveLength(1); defined(gates[0]).resolve({ report: "one done" }); defined(gates[1]).resolve({ report: "two done" }); }); + + test("three concurrent mutating shared-cwd lanes log one concurrent-lane-overlap", async () => { + const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-wave-")); + const gates = [ + deferred(), + deferred(), + deferred(), + ]; + let callIndex = 0; + const deps = makeDeps(async () => defined(gates[callIndex++]).promise, { + cwd: "/repo", + }); + deps.getWorkdirBase = () => dir; + const spawn = createSpawnAgentTool(deps); + + for (const label of ["one", "two", "three"] as const) { + await callTool(spawn, { + description: `build ${label}`, + prompt: `implement thing ${label}`, + intent: "implement", + success_criteria: [`thing ${label} ships`], + }); + } + + const path = join(dir, INTERVENTION_FILE); + let log = ""; + for (let i = 0; i < 50; i++) { + try { + log = await readFile(path, "utf8"); + if (log.includes("concurrent-lane-overlap")) break; + } catch { + // append is fire-and-forget + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + expect( + log + .trim() + .split("\n") + .filter((line) => line.includes("concurrent-lane-overlap")), + ).toHaveLength(1); + + for (const gate of gates) defined(gate).resolve({ report: "done" }); + }); + + test("shared-cwd explore then implement does not log concurrent-lane-overlap", async () => { + const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-readonly-")); + const gates = [ + deferred(), + deferred(), + ]; + let callIndex = 0; + const deps = makeDeps(async () => defined(gates[callIndex++]).promise, { + cwd: "/repo", + }); + deps.getWorkdirBase = () => dir; + const spawn = createSpawnAgentTool(deps); + + await callTool(spawn, { + description: "look around", + prompt: "map the tree", + intent: "explore", + }); + await callTool(spawn, { + description: "build one", + prompt: "implement thing one", + intent: "implement", + success_criteria: ["thing one ships"], + }); + await new Promise((resolve) => setTimeout(resolve, 50)); + + await expect( + readFile(join(dir, INTERVENTION_FILE), "utf8"), + ).rejects.toThrow(); + + defined(gates[0]).resolve({ report: "mapped" }); + defined(gates[1]).resolve({ report: "one done" }); + }); + + test("a later mutating wave can warn again after the prior wave settles", async () => { + const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-reset-")); + const gates = [ + deferred(), + deferred(), + deferred(), + deferred(), + ]; + let callIndex = 0; + const deps = makeDeps(async () => defined(gates[callIndex++]).promise, { + cwd: "/repo", + }); + deps.getWorkdirBase = () => dir; + const spawn = createSpawnAgentTool(deps); + + await callTool(spawn, { + description: "wave1 a", + prompt: "implement a", + intent: "implement", + success_criteria: ["a ships"], + }); + await callTool(spawn, { + description: "wave1 b", + prompt: "implement b", + intent: "implement", + success_criteria: ["b ships"], + }); + + const path = join(dir, INTERVENTION_FILE); + let log = ""; + for (let i = 0; i < 50; i++) { + try { + log = await readFile(path, "utf8"); + if (log.includes("concurrent-lane-overlap")) break; + } catch { + // append is fire-and-forget + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + expect( + log + .trim() + .split("\n") + .filter((line) => line.includes("concurrent-lane-overlap")), + ).toHaveLength(1); + + defined(gates[0]).resolve({ report: "a done" }); + defined(gates[1]).resolve({ report: "b done" }); + await new Promise((resolve) => setTimeout(resolve, 30)); + + await callTool(spawn, { + description: "wave2 a", + prompt: "implement c", + intent: "implement", + success_criteria: ["c ships"], + }); + await callTool(spawn, { + description: "wave2 b", + prompt: "implement d", + intent: "implement", + success_criteria: ["d ships"], + }); + + for (let i = 0; i < 50; i++) { + try { + log = await readFile(path, "utf8"); + if ( + log + .trim() + .split("\n") + .filter((line) => line.includes("concurrent-lane-overlap")) + .length >= 2 + ) { + break; + } + } catch { + // append is fire-and-forget + } + await new Promise((resolve) => setTimeout(resolve, 20)); + } + expect( + log + .trim() + .split("\n") + .filter((line) => line.includes("concurrent-lane-overlap")), + ).toHaveLength(2); + expect(log).toContain("wave2"); + + defined(gates[2]).resolve({ report: "c done" }); + defined(gates[3]).resolve({ report: "d done" }); + }); + + test("a queued mutating peer does not log concurrent-lane-overlap", async () => { + const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-queued-")); + const gates = [ + deferred(), + deferred(), + ]; + let callIndex = 0; + const deps = makeDeps(async () => defined(gates[callIndex++]).promise, { + cwd: "/repo", + }); + deps.getWorkdirBase = () => dir; + deps.admission = createAdmissionQueue({ capacity: 1 }); + const spawn = createSpawnAgentTool(deps); + + const first = await callTool(spawn, { + description: "holder", + prompt: "implement holder", + intent: "implement", + success_criteria: ["holder ships"], + }); + const queued = await callTool(spawn, { + description: "queued writer", + prompt: "implement queued", + intent: "implement", + success_criteria: ["queued ships"], + }); + expect(first.status).toBe("running"); + expect(queued.status).toBe("queued"); + await new Promise((resolve) => setTimeout(resolve, 50)); + + await expect( + readFile(join(dir, INTERVENTION_FILE), "utf8"), + ).rejects.toThrow(); + + defined(gates[0]).resolve({ report: "holder done" }); + defined(gates[1]).resolve({ report: "queued done" }); + }); }); describe("wait mailbox session tombstone and pin", () => { diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 936ecbe01..42aeedefe 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -92,7 +92,7 @@ import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js"; import { classifyAgentName } from "../telemetry/classify.js"; import { captureSubagentEnd } from "../telemetry/product-events.js"; import { getCurrentTurnTraceId } from "../telemetry/feedback.js"; -import type { DirectorPackage } from "../agent/directors/types.js"; +import type { DirectorPackage, ModelRole } from "../agent/directors/types.js"; import { SPAWN_AGENT_TOOL_NAME } from "./tool-taxonomy.js"; import { assertCanTargetAgent, @@ -868,11 +868,19 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { // will run in — worktree-isolated lanes always get a fresh, disjoint path // here, so this can only ever fire in the shared-cwd fallback, which is // exactly where two lanes really can stomp each other's writes. - // Keyed by call.id so a completed lane (removed when the worker settles) - // is never mistaken for one still running: sequential dispatches to the - // same cwd are always clean. Tracking lasts the worker lifetime, not the - // immediate spawn_agent return. - const activeLanes = new Map(); + // Keyed by call.id for finally cleanup. The session store is authoritative + // for liveness: cancel (and other terminals) can stamp finishedAt before the + // run promise settles and reaches finally, so a map entry alone is not proof + // the lane is still working. + // Warnings are for mutating cwd waves only: declared read-only modelRoles + // (explore/plan/review/test) never participate, and a cwd emits at most one + // conflict while any live mutating writer remains; the wave flag clears when + // that set empties so a later independent wave can warn once again. + const activeLanes = new Map< + string, + { description: string; cwd: string; modelRole: ModelRole | undefined } + >(); + const warnedMutatingCwds = new Set(); let conflictLog: InterventionSink | null = null; const recordConflict = (event: Parameters[0]): void => { conflictLog ??= createInterventionLog(deps.getWorkdirBase(), { @@ -880,6 +888,35 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }); conflictLog(event); }; + const isOverlapLive = (id: string): boolean => { + const session = deps.sessions.get(id); + return ( + session !== undefined && + (session.lifecycle.state === "pending_init" || + session.lifecycle.state === "running") + ); + }; + const isDeclaredReadOnly = (modelRole: ModelRole | undefined): boolean => + modelRole === "explore" || + modelRole === "plan" || + modelRole === "review" || + modelRole === "test"; + const clearWarnedCwdsWithoutLiveWriters = (): void => { + const liveWriterCwds = new Set(); + for (const [id, lane] of activeLanes) { + if (!isOverlapLive(id)) { + activeLanes.delete(id); + continue; + } + if (!isDeclaredReadOnly(lane.modelRole)) { + liveWriterCwds.add(lane.cwd); + } + } + for (const cwd of [...warnedMutatingCwds]) { + if (!liveWriterCwds.has(cwd)) warnedMutatingCwds.delete(cwd); + } + }; + return tool({ definition: spawnAgentToolDefinition, handler: async (call, _signal): Promise => { @@ -1234,26 +1271,51 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { return; } - // Detect, don't lock: warn when another lane still running right now - // is already working in this same cwd. Worktree-isolated lanes never - // collide here (each gets its own directory); this only fires in the - // shared-cwd fallback, where two lanes genuinely can overwrite each - // other's writes. Never blocks the spawn — the least destructive + // Detect, don't lock: warn when another live mutating lane is already + // working in this same cwd. Worktree-isolated lanes never collide here + // (each gets its own directory); this only fires in the shared-cwd + // fallback, where two writers genuinely can overwrite each other's + // writes. Declared read-only modelRoles are ignored. At most one + // conflict per cwd wave. Never blocks the spawn — the least destructive // response that still tells the operator something true, since a // shared cwd does not by itself prove the two lanes touch the same // files, only that they could. const laneCwd = worktreeCwd ?? deps.cwd; + const laneModelRole = resolved.pkg?.modelRole; + const laneIsWriter = !isDeclaredReadOnly(laneModelRole); + clearWarnedCwdsWithoutLiveWriters(); + let liveMutatingPeer: { id: string; description: string } | undefined; for (const [otherId, other] of activeLanes) { + if (!isOverlapLive(otherId)) { + activeLanes.delete(otherId); + continue; + } if (other.cwd !== laneCwd) continue; + if (isDeclaredReadOnly(other.modelRole)) continue; + liveMutatingPeer ??= { + id: otherId, + description: other.description, + }; + } + if ( + laneIsWriter && + liveMutatingPeer !== undefined && + !warnedMutatingCwds.has(laneCwd) + ) { + warnedMutatingCwds.add(laneCwd); recordConflict({ id: "concurrent-lane-overlap", class: "conflict", detail: - `"${description}" (${call.id}) and "${other.description}" (${otherId}) ` + + `"${description}" (${call.id}) and "${liveMutatingPeer.description}" (${liveMutatingPeer.id}) ` + `are both running against ${laneCwd} at once`, }); } - activeLanes.set(call.id, { description, cwd: laneCwd }); + activeLanes.set(call.id, { + description, + cwd: laneCwd, + modelRole: laneModelRole, + }); const params: RunSubAgentParams = { // Name the trace directory after the session-store id so the @@ -1449,6 +1511,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool { }) .finally(() => { activeLanes.delete(call.id); + clearWarnedCwdsWithoutLiveWriters(); finalizeEnd(); if (!keepWorktreeAlive) void reclaimWorktree(); admission.release(session.id);