Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion docs/IMPLEMENTATION.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
253 changes: 253 additions & 0 deletions src/subagent/agent-fleet.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down Expand Up @@ -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", () => {
Expand Down
Loading
Loading