Rate-limit concurrent mutating lane-overlap warnings (#891)

* Gate lane-overlap warnings on live session-store status

Cancel can stamp finishedAt before the run promise settles and
reaches finally, so an activeLanes map entry alone was enough to
emit concurrent-lane-overlap against a worker that was already
terminal. Overlap checks now trust pending_init/running from the
session store, prune terminal or absent map entries opportunistically,
and keep the finally delete.

* Rate-limit mutating cwd overlap warnings and skip read-only peers

Shared-cwd concurrent-lane-overlap noise was one record per peer with
no filter for explore/plan/review/test directors. Emit at most one
conflict when a cwd wave first has two live mutating writers, suppress
declared read-only peers, and clear the wave flag once no live writer
remains. No age TTL or map sweep.

* Cover queued writers in concurrent-lane-overlap tests

Author: TheGreatAxios

docs/IMPLEMENTATION.md

@@ -167,7 +167,7 @@ Sixteen packages under `src/agent/directors/<id>/` 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 `<env>` injects cwd, platform, arch, runtime, date, and git status on every chat and worker prompt.

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<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+    ];
+    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<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+    ];
+    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<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+    ];
+    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<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+    ];
+    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<RunSubAgentResult>(),
+      deferred<RunSubAgentResult>(),
+    ];
+    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", () => {