Skip to content

Commit 8610d41

Browse files
TheGreatAxiosTheGreatAxios
committed
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.
1 parent 68a55f3 commit 8610d41

3 files changed

Lines changed: 246 additions & 16 deletions

File tree

‎docs/IMPLEMENTATION.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -167,7 +167,7 @@ Sixteen packages under `src/agent/directors/<id>/` register in `DIRECTOR_REGISTR
167167

168168
**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.
169169

170-
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 lane whose session-store lifecycle is still `pending_init` or `running` records a `concurrent-lane-overlap` entry in `intervention-log.ts` (class `conflict`). 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.
170+
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.
171171
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.
172172

173173
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‎

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -745,10 +745,186 @@ describe("spawn_agent same-cwd concurrency", () => {
745745
expect(log).toContain("/repo");
746746
expect(log).toContain("build one");
747747
expect(log).toContain("build two");
748+
expect(
749+
log
750+
.trim()
751+
.split("\n")
752+
.filter((line) => line.includes("concurrent-lane-overlap")),
753+
).toHaveLength(1);
748754

749755
defined(gates[0]).resolve({ report: "one done" });
750756
defined(gates[1]).resolve({ report: "two done" });
751757
});
758+
759+
test("three concurrent mutating shared-cwd lanes log one concurrent-lane-overlap", async () => {
760+
const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-wave-"));
761+
const gates = [
762+
deferred<RunSubAgentResult>(),
763+
deferred<RunSubAgentResult>(),
764+
deferred<RunSubAgentResult>(),
765+
];
766+
let callIndex = 0;
767+
const deps = makeDeps(async () => defined(gates[callIndex++]).promise, {
768+
cwd: "/repo",
769+
});
770+
deps.getWorkdirBase = () => dir;
771+
const spawn = createSpawnAgentTool(deps);
772+
773+
for (const label of ["one", "two", "three"] as const) {
774+
await callTool(spawn, {
775+
description: `build ${label}`,
776+
prompt: `implement thing ${label}`,
777+
intent: "implement",
778+
success_criteria: [`thing ${label} ships`],
779+
});
780+
}
781+
782+
const path = join(dir, INTERVENTION_FILE);
783+
let log = "";
784+
for (let i = 0; i < 50; i++) {
785+
try {
786+
log = await readFile(path, "utf8");
787+
if (log.includes("concurrent-lane-overlap")) break;
788+
} catch {
789+
// append is fire-and-forget
790+
}
791+
await new Promise((resolve) => setTimeout(resolve, 20));
792+
}
793+
expect(
794+
log
795+
.trim()
796+
.split("\n")
797+
.filter((line) => line.includes("concurrent-lane-overlap")),
798+
).toHaveLength(1);
799+
800+
for (const gate of gates) defined(gate).resolve({ report: "done" });
801+
});
802+
803+
test("shared-cwd explore then implement does not log concurrent-lane-overlap", async () => {
804+
const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-readonly-"));
805+
const gates = [
806+
deferred<RunSubAgentResult>(),
807+
deferred<RunSubAgentResult>(),
808+
];
809+
let callIndex = 0;
810+
const deps = makeDeps(async () => defined(gates[callIndex++]).promise, {
811+
cwd: "/repo",
812+
});
813+
deps.getWorkdirBase = () => dir;
814+
const spawn = createSpawnAgentTool(deps);
815+
816+
await callTool(spawn, {
817+
description: "look around",
818+
prompt: "map the tree",
819+
intent: "explore",
820+
});
821+
await callTool(spawn, {
822+
description: "build one",
823+
prompt: "implement thing one",
824+
intent: "implement",
825+
success_criteria: ["thing one ships"],
826+
});
827+
await new Promise((resolve) => setTimeout(resolve, 50));
828+
829+
await expect(
830+
readFile(join(dir, INTERVENTION_FILE), "utf8"),
831+
).rejects.toThrow();
832+
833+
defined(gates[0]).resolve({ report: "mapped" });
834+
defined(gates[1]).resolve({ report: "one done" });
835+
});
836+
837+
test("a later mutating wave can warn again after the prior wave settles", async () => {
838+
const dir = await mkdtemp(join(tmpdir(), "fleet-overlap-reset-"));
839+
const gates = [
840+
deferred<RunSubAgentResult>(),
841+
deferred<RunSubAgentResult>(),
842+
deferred<RunSubAgentResult>(),
843+
deferred<RunSubAgentResult>(),
844+
];
845+
let callIndex = 0;
846+
const deps = makeDeps(async () => defined(gates[callIndex++]).promise, {
847+
cwd: "/repo",
848+
});
849+
deps.getWorkdirBase = () => dir;
850+
const spawn = createSpawnAgentTool(deps);
851+
852+
await callTool(spawn, {
853+
description: "wave1 a",
854+
prompt: "implement a",
855+
intent: "implement",
856+
success_criteria: ["a ships"],
857+
});
858+
await callTool(spawn, {
859+
description: "wave1 b",
860+
prompt: "implement b",
861+
intent: "implement",
862+
success_criteria: ["b ships"],
863+
});
864+
865+
const path = join(dir, INTERVENTION_FILE);
866+
let log = "";
867+
for (let i = 0; i < 50; i++) {
868+
try {
869+
log = await readFile(path, "utf8");
870+
if (log.includes("concurrent-lane-overlap")) break;
871+
} catch {
872+
// append is fire-and-forget
873+
}
874+
await new Promise((resolve) => setTimeout(resolve, 20));
875+
}
876+
expect(
877+
log
878+
.trim()
879+
.split("\n")
880+
.filter((line) => line.includes("concurrent-lane-overlap")),
881+
).toHaveLength(1);
882+
883+
defined(gates[0]).resolve({ report: "a done" });
884+
defined(gates[1]).resolve({ report: "b done" });
885+
await new Promise((resolve) => setTimeout(resolve, 30));
886+
887+
await callTool(spawn, {
888+
description: "wave2 a",
889+
prompt: "implement c",
890+
intent: "implement",
891+
success_criteria: ["c ships"],
892+
});
893+
await callTool(spawn, {
894+
description: "wave2 b",
895+
prompt: "implement d",
896+
intent: "implement",
897+
success_criteria: ["d ships"],
898+
});
899+
900+
for (let i = 0; i < 50; i++) {
901+
try {
902+
log = await readFile(path, "utf8");
903+
if (
904+
log
905+
.trim()
906+
.split("\n")
907+
.filter((line) => line.includes("concurrent-lane-overlap"))
908+
.length >= 2
909+
) {
910+
break;
911+
}
912+
} catch {
913+
// append is fire-and-forget
914+
}
915+
await new Promise((resolve) => setTimeout(resolve, 20));
916+
}
917+
expect(
918+
log
919+
.trim()
920+
.split("\n")
921+
.filter((line) => line.includes("concurrent-lane-overlap")),
922+
).toHaveLength(2);
923+
expect(log).toContain("wave2");
924+
925+
defined(gates[2]).resolve({ report: "c done" });
926+
defined(gates[3]).resolve({ report: "d done" });
927+
});
752928
});
753929

754930
describe("wait mailbox session tombstone and pin", () => {

‎src/subagent/agent-fleet.ts‎

Lines changed: 69 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ import { NOOP_TELEMETRY, type Telemetry } from "../telemetry/index.js";
9292
import { classifyAgentName } from "../telemetry/classify.js";
9393
import { captureSubagentEnd } from "../telemetry/product-events.js";
9494
import { getCurrentTurnTraceId } from "../telemetry/feedback.js";
95-
import type { DirectorPackage } from "../agent/directors/types.js";
95+
import type { DirectorPackage, ModelRole } from "../agent/directors/types.js";
9696
import { SPAWN_AGENT_TOOL_NAME } from "./tool-taxonomy.js";
9797
import {
9898
assertCanTargetAgent,
@@ -872,14 +872,51 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
872872
// for liveness: cancel (and other terminals) can stamp finishedAt before the
873873
// run promise settles and reaches finally, so a map entry alone is not proof
874874
// the lane is still working.
875-
const activeLanes = new Map<string, { description: string; cwd: string }>();
875+
// Warnings are for mutating cwd waves only: declared read-only modelRoles
876+
// (explore/plan/review/test) never participate, and a cwd emits at most one
877+
// conflict while any live mutating writer remains; the wave flag clears when
878+
// that set empties so a later independent wave can warn once again.
879+
const activeLanes = new Map<
880+
string,
881+
{ description: string; cwd: string; modelRole: ModelRole | undefined }
882+
>();
883+
const warnedMutatingCwds = new Set<string>();
876884
let conflictLog: InterventionSink | null = null;
877885
const recordConflict = (event: Parameters<InterventionSink>[0]): void => {
878886
conflictLog ??= createInterventionLog(deps.getWorkdirBase(), {
879887
role: "parent",
880888
});
881889
conflictLog(event);
882890
};
891+
const isOverlapLive = (id: string): boolean => {
892+
const session = deps.sessions.get(id);
893+
return (
894+
session !== undefined &&
895+
(session.lifecycle.state === "pending_init" ||
896+
session.lifecycle.state === "running")
897+
);
898+
};
899+
const isDeclaredReadOnly = (modelRole: ModelRole | undefined): boolean =>
900+
modelRole === "explore" ||
901+
modelRole === "plan" ||
902+
modelRole === "review" ||
903+
modelRole === "test";
904+
const clearWarnedCwdsWithoutLiveWriters = (): void => {
905+
const liveWriterCwds = new Set<string>();
906+
for (const [id, lane] of activeLanes) {
907+
if (!isOverlapLive(id)) {
908+
activeLanes.delete(id);
909+
continue;
910+
}
911+
if (!isDeclaredReadOnly(lane.modelRole)) {
912+
liveWriterCwds.add(lane.cwd);
913+
}
914+
}
915+
for (const cwd of [...warnedMutatingCwds]) {
916+
if (!liveWriterCwds.has(cwd)) warnedMutatingCwds.delete(cwd);
917+
}
918+
};
919+
883920
return tool({
884921
definition: spawnAgentToolDefinition,
885922
handler: async (call, _signal): Promise<ToolResult> => {
@@ -1234,35 +1271,51 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
12341271
return;
12351272
}
12361273

1237-
// Detect, don't lock: warn when another lane still running right now
1238-
// is already working in this same cwd. Worktree-isolated lanes never
1239-
// collide here (each gets its own directory); this only fires in the
1240-
// shared-cwd fallback, where two lanes genuinely can overwrite each
1241-
// other's writes. Never blocks the spawn — the least destructive
1274+
// Detect, don't lock: warn when another live mutating lane is already
1275+
// working in this same cwd. Worktree-isolated lanes never collide here
1276+
// (each gets its own directory); this only fires in the shared-cwd
1277+
// fallback, where two writers genuinely can overwrite each other's
1278+
// writes. Declared read-only modelRoles are ignored. At most one
1279+
// conflict per cwd wave. Never blocks the spawn — the least destructive
12421280
// response that still tells the operator something true, since a
12431281
// shared cwd does not by itself prove the two lanes touch the same
12441282
// files, only that they could.
12451283
const laneCwd = worktreeCwd ?? deps.cwd;
1284+
const laneModelRole = resolved.pkg?.modelRole;
1285+
const laneIsWriter = !isDeclaredReadOnly(laneModelRole);
1286+
clearWarnedCwdsWithoutLiveWriters();
1287+
let liveMutatingPeer: { id: string; description: string } | undefined;
12461288
for (const [otherId, other] of activeLanes) {
1247-
const otherSession = deps.sessions.get(otherId);
1248-
if (
1249-
otherSession === undefined ||
1250-
(otherSession.lifecycle.state !== "pending_init" &&
1251-
otherSession.lifecycle.state !== "running")
1252-
) {
1289+
if (!isOverlapLive(otherId)) {
12531290
activeLanes.delete(otherId);
12541291
continue;
12551292
}
12561293
if (other.cwd !== laneCwd) continue;
1294+
if (isDeclaredReadOnly(other.modelRole)) continue;
1295+
liveMutatingPeer ??= {
1296+
id: otherId,
1297+
description: other.description,
1298+
};
1299+
}
1300+
if (
1301+
laneIsWriter &&
1302+
liveMutatingPeer !== undefined &&
1303+
!warnedMutatingCwds.has(laneCwd)
1304+
) {
1305+
warnedMutatingCwds.add(laneCwd);
12571306
recordConflict({
12581307
id: "concurrent-lane-overlap",
12591308
class: "conflict",
12601309
detail:
1261-
`"${description}" (${call.id}) and "${other.description}" (${otherId}) ` +
1310+
`"${description}" (${call.id}) and "${liveMutatingPeer.description}" (${liveMutatingPeer.id}) ` +
12621311
`are both running against ${laneCwd} at once`,
12631312
});
12641313
}
1265-
activeLanes.set(call.id, { description, cwd: laneCwd });
1314+
activeLanes.set(call.id, {
1315+
description,
1316+
cwd: laneCwd,
1317+
modelRole: laneModelRole,
1318+
});
12661319

12671320
const params: RunSubAgentParams = {
12681321
// Name the trace directory after the session-store id so the
@@ -1458,6 +1511,7 @@ export function createSpawnAgentTool(deps: AgentFleetDeps): AgentTool {
14581511
})
14591512
.finally(() => {
14601513
activeLanes.delete(call.id);
1514+
clearWarnedCwdsWithoutLiveWriters();
14611515
finalizeEnd();
14621516
if (!keepWorktreeAlive) void reclaimWorktree();
14631517
admission.release(session.id);

0 commit comments

Comments
 (0)