rock/homelab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(agent-pod): persistent per-role agent pool, concurrency moves to repo level

Browse Source 
coordinator.js now runs one long-lived planner/investigator/implementer/judge
session per repo (reused across every task via tmux send-keys) instead of a
fresh spawn per task per stage. Tasks within a repo run sequentially against
that pool; concurrency is now REPO_CONCURRENCY (default 3) concurrent repos
via a new --repos flag, not concurrent tasks in one repo's phase.
This commit is contained in:
Story Crater Bot
2026-08-18 21:30:49 -07:00
parent 5a8fc1885a
commit 35bef19e0c
2 changed files with 221 additions and 342 deletions
Download Patch File Download Diff File Expand all files Collapse all files
k8s/apps/agent-pod/coordinator-configmap.yaml
+202 -325
View File
@@ -8,20 +8,35 @@ data:
#!/usr/bin/env node #!/usr/bin/env node
// coordinator: local CLI that drives a multi-phase, multi-task pipeline of // coordinator: local CLI that drives a multi-phase, multi-task pipeline of
// planner/investigator/implementer/judge stages, same state machine as // planner/investigator/implementer/judge stages, same state machine as
// hub.js's old /pipeline handler. The difference is who owns each stage's // hub.js's old /pipeline handler. Every interactive stage runs through the
// session: hub.js spawned raw tmux sessions itself and polled sentinel // patched agent-manager fork's `spawn` subcommand, so it is a real tracked
// files on a 10s timer to notice completion. This spawns every interactive // session (tmux pane on agent-manager's private socket + a state.db row)
// stage through the patched agent-manager fork's `spawn` subcommand, so it // from the moment it exists -- attachable and visible in agent-manager's
// is a real tracked session (tmux pane on agent-manager's private socket + // own TUI the whole time it runs.
// a state.db row) from the moment it exists -- attachable and visible in //
// agent-manager's own TUI the whole time it runs. The coordinator never // Per repo, each role (planner/investigator/implementer/judge) is ONE
// kills a stage's session; it rests at an idle prompt once its sentinel // persistent agent-manager session, not a fresh spawn per task: the first
// file lands, and agent-manager's session list becomes the audit trail of // task to need a role spawns it, every later task for that role reuses the
// everything the pipeline ran. Completion is still signaled by sentinel // same tmux pane via `tmux send-keys` (see runOnPool) -- the same nudge
// files under the task's worktree (unchanged convention), but waited on // mechanism that used to only fire on a stall now doubles as "give this
// agent its next task." A role's conversation history accumulates across
// every task it ever handles for that repo; nothing resets it mid-pipeline.
// Because only one implementer/judge/etc. exists per repo, tasks within a
// phase run strictly sequentially against the pool -- no per-task worktree,
// no per-task branch, no merge-back step; every task commits straight onto
// the phase branch in the repo's one shared clone.
//
// The unit of concurrency is now the REPO, not the task: runCoordinator
// takes a list of repos and runs up to REPO_CONCURRENCY of them at once,
// each with its own clone (under WORK_DIR/<repoId>) and its own 4-agent
// pool. The coordinator never kills a role's session; it rests at an idle
// prompt between tasks, and agent-manager's session list becomes the audit
// trail of everything every repo's pipeline ran. Completion is signaled by
// sentinel files under the repo's clone (unchanged convention), waited on
// with fs.watch instead of polling. // with fs.watch instead of polling.
const fs = require("node:fs"); const fs = require("node:fs");
const path = require("node:path"); const path = require("node:path");
const crypto = require("node:crypto");
const { spawn } = require("node:child_process"); const { spawn } = require("node:child_process");
const WORK_DIR = process.env.HUB_WORK_DIR || path.join(require("node:os").tmpdir(), "agent-harness-work"); const WORK_DIR = process.env.HUB_WORK_DIR || path.join(require("node:os").tmpdir(), "agent-harness-work");
@@ -37,31 +52,29 @@ data:
// Empty means "let pi fall back to ~/.pi/agent/settings.json's default" // Empty means "let pi fall back to ~/.pi/agent/settings.json's default"
// (currently anthropic/claude-sonnet-4-5, real paid usage). Set both to // (currently anthropic/claude-sonnet-4-5, real paid usage). Set both to
// route every stage -- headless (spawnPi) and interactive (runStage) -- // route every stage -- headless (spawnPi) and interactive (runOnPool) --
// at the homelab model instead, e.g. AGENT_PROVIDER=homelab-ornith // at the homelab model instead, e.g. AGENT_PROVIDER=homelab-ornith
// AGENT_MODEL=ornith:35b. // AGENT_MODEL=ornith:35b.
const AGENT_PROVIDER = process.env.AGENT_PROVIDER || ""; const AGENT_PROVIDER = process.env.AGENT_PROVIDER || "";
const AGENT_MODEL = process.env.AGENT_MODEL || ""; const AGENT_MODEL = process.env.AGENT_MODEL || "";
// judge (both per-task and phase-judge -- both spawn with stageLabel // judge can run a different model than the rest of the chain, e.g.
// "judge", see runTaskInteractive/runPipeline) can run a different model // homelab-reasoning instead of homelab-ornith now that verifier/PRM is
// than the rest of the chain, e.g. homelab-reasoning instead of // retired. Falls back to AGENT_PROVIDER/AGENT_MODEL when unset, so a run
// homelab-ornith now that verifier/PRM is retired. Falls back to // that doesn't care keeps one uniform model everywhere.
// AGENT_PROVIDER/AGENT_MODEL when unset, so a run that doesn't care keeps
// one uniform model everywhere, same as before this existed.
const JUDGE_PROVIDER = process.env.JUDGE_PROVIDER || AGENT_PROVIDER; const JUDGE_PROVIDER = process.env.JUDGE_PROVIDER || AGENT_PROVIDER;
const JUDGE_MODEL = process.env.JUDGE_MODEL || AGENT_MODEL; const JUDGE_MODEL = process.env.JUDGE_MODEL || AGENT_MODEL;
function providerModelFor(stageLabel) { function providerModelFor(role) {
return stageLabel === "judge" ? { provider: JUDGE_PROVIDER, model: JUDGE_MODEL } : { provider: AGENT_PROVIDER, model: AGENT_MODEL }; return role === "judge" ? { provider: JUDGE_PROVIDER, model: JUDGE_MODEL } : { provider: AGENT_PROVIDER, model: AGENT_MODEL };
} }
// agent-manager's private tmux server and session-naming scheme // agent-manager's private tmux server and session-naming scheme
// (internal/tmux/tmux.go: defaultSocket = "agentmgr", sessionName(id) = // (internal/tmux/tmux.go: defaultSocket = "agentmgr", sessionName(id) =
// "am_"+id) -- stable, documented internals of the fork, used here only // "am_"+id) -- stable, documented internals of the fork, used here only
// for read-only introspection (pane capture) and the RETRY nudge, exactly // for read-only introspection (pane capture) and role nudges, exactly the
// the class of operation hub.js already ran directly against its own // class of operation hub.js already ran directly against its own sessions
// sessions rather than asking a model to do it. // rather than asking a model to do it.
const AM_SOCKET = "agentmgr"; const AM_SOCKET = "agentmgr";
function amSessionName(id) { function amSessionName(id) {
return `am_${id}`; return `am_${id}`;
@@ -112,8 +125,8 @@ data:
// Headless one-shot pi call (`pi -p --mode json <prompt>`), used only for // Headless one-shot pi call (`pi -p --mode json <prompt>`), used only for
// quick diagnostic/mechanical calls that don't need to be a watchable // quick diagnostic/mechanical calls that don't need to be a watchable
// session: resolver's crash/stall diagnosis, and the initial clone. Kept // session: resolver's crash/stall diagnosis, and the initial clone. Kept
// exactly as before -- only the interactive per-role stages (runStage, // exactly as before -- only the interactive per-role stages (runOnPool,
// below) move to agent-manager. // below) go through agent-manager.
function spawnPi({ agent, prompt, cwd }) { function spawnPi({ agent, prompt, cwd }) {
const finalPrompt = ROLE_SKILLS.has(agent) ? `/skill:${agent} ${HARD_RULES}${prompt}` : prompt; const finalPrompt = ROLE_SKILLS.has(agent) ? `/skill:${agent} ${HARD_RULES}${prompt}` : prompt;
const args = ["-p", "--mode", "json"]; const args = ["-p", "--mode", "json"];
@@ -154,99 +167,17 @@ data:
}); });
} }
async function askResolver(cwd, task, diagnosticPrompt) { async function askResolver(cwd, repoId, diagnosticPrompt) {
const result = await spawnPi({ agent: "resolver", prompt: diagnosticPrompt, cwd }); const result = await spawnPi({ agent: "resolver", prompt: diagnosticPrompt, cwd });
return parseVerdictLine(result.lastText, "RESOLUTION"); return parseVerdictLine(result.lastText, "RESOLUTION");
} }
function taskSessionLabel(task) {
return `task-${task.replace(/[^a-zA-Z0-9]/g, "-")}`;
}
// POSIX single-quote wrap, for embedding a literal `agent-manager spawn`
// command inside another stage's prompt -- the same escaping tmux.ShellQuote
// does on the Go side of the fork.
function shQuote(s) {
return "'" + String(s).replace(/'/g, `'\\''`) + "'";
}
// The literal shell command an agent runs, as its own last action, to spawn
// its successor stage itself instead of the coordinator spawning it. Always
// applies HARD_RULES to the embedded prompt -- the single place that
// happens for agent-issued spawns, mirroring spawnRole's HARD_RULES + prompt
// for coordinator-issued ones (see runTaskInteractive) so it's applied
// exactly once regardless of who calls agent-manager spawn.
function spawnCommandLine(cwd, task, stageLabel, prompt) {
const label = `${taskSessionLabel(task)}-${stageLabel}`;
const parts = [AGENT_MANAGER_BIN, "spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--prompt", HARD_RULES + prompt];
const { provider, model } = providerModelFor(stageLabel);
if (provider) parts.push("--provider", provider);
if (model) parts.push("--model", model);
return parts.map(shQuote).join(" ");
}
// Instruction builders for the self-chained happy path: planner spawns
// investigator spawns implementer spawns judge, each as its own last
// action. Built bottom-up (judge first) since each level embeds the next
// level's full spawn command as literal text. Judge never self-spawns --
// the coordinator reads its verdict and decides PASS/retry/revise itself
// (see runTaskInteractive), so retries and plan revisions stay a
// deterministic, hard-capped branch in code, not something an agent
// decides on its own.
function judgeInstructions(task, baseBranch, attempt) {
return (
`Use the judge skill to review the diff against ${baseBranch}...HEAD. Write your verdict to ` +
`.task-result-${task} as a single "VERDICT: PASS" or "VERDICT: FAIL" line plus one line of ` +
`rationale, then run: touch .stage-done-${task}-judge-${attempt}`
);
}
function implementerInstructions(cwd, task, baseBranch, attempt, feedbackHint) {
const judgeSpawn = spawnCommandLine(cwd, task, "judge", judgeInstructions(task, baseBranch, attempt));
return (
`Use the implementer skill to implement what the current PLAN.md specifies (commit as you go). ${feedbackHint} ` +
`When your implementation is committed, spawn the judge stage yourself by running exactly this command: ${judgeSpawn}\n\n` +
`Then run: touch .stage-done-${task}-implementer-${attempt}`
);
}
function investigatorInstructions(cwd, task, baseBranch) {
const implSpawn = spawnCommandLine(cwd, task, "implementer", implementerInstructions(cwd, task, baseBranch, 1, ""));
return (
`Use the investigator skill to confirm PLAN.md against real sources, append findings, commit. ` +
`When finished, spawn the implementer stage yourself by running exactly this command: ${implSpawn}\n\n` +
`Then run: touch .stage-done-${task}-investigator`
);
}
function plannerInstructions(cwd, task, baseBranch, specHint) {
const invSpawn = spawnCommandLine(cwd, task, "investigator", investigatorInstructions(cwd, task, baseBranch));
return (
`Use the planner skill to draft PLAN.md for task ${task}, reading its spec (${specHint}). Commit PLAN.md. ` +
`When finished, spawn the investigator stage yourself by running exactly this command: ${invSpawn}\n\n` +
`Then run: touch .stage-done-${task}-planner`
);
}
async function runConcurrent(items, limit, worker) {
const results = new Array(items.length);
let i = 0;
async function next() {
while (i < items.length) {
const idx = i++;
results[idx] = await worker(items[idx], idx);
}
}
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, next));
return results;
}
const STAGE_TIMEOUT_MS = 10 * 60 * 1000; const STAGE_TIMEOUT_MS = 10 * 60 * 1000;
const NUDGE_TIMEOUT_MS = 5 * 60 * 1000; const NUDGE_TIMEOUT_MS = 5 * 60 * 1000;
// Resolves as soon as filePath appears (fs.watch on its directory), or // Resolves as soon as filePath appears (fs.watch on its directory), or
// after limitMs with no sign of it -- the "notify" half of the migration: // after limitMs with no sign of it -- event-driven completion instead of
// event-driven completion instead of hub.js's old 10s poll. // hub.js's old 10s poll.
function waitForSentinel(filePath, limitMs) { function waitForSentinel(filePath, limitMs) {
return new Promise((resolve) => { return new Promise((resolve) => {
if (fs.existsSync(filePath)) return resolve(true); if (fs.existsSync(filePath)) return resolve(true);
@@ -282,119 +213,104 @@ data:
}); });
} }
// Runs one role as its own fresh interactive pi session, spawned as a real // Runs one task's worth of work on a persistent per-role agent: spawns the
// agent-manager-tracked session (see AGENT_MANAGER_BIN above) instead of a // role's session the first time it's ever needed for this repo, sends every
// raw tmux pane. Planner, investigator, implementer, judge, and // later prompt into that same tmux pane via send-keys. pool is a plain
// phase-judge are separate agents with separate context, coordinating only // object keyed by role name ("planner"/"investigator"/"implementer"/
// through what's on disk in the task's worktree -- unchanged. The session // "judge"), shared across every task in a repo's pipeline (see
// is never killed here: it rests at an idle prompt once its sentinel file // runRepoPipeline) -- it IS the 4-agent pool, one entry per role, filled in
// lands, staying attachable in agent-manager for as long as the user wants. // lazily as each role gets its first task.
// async function runOnPool(pool, cwd, repoId, role, prompt, sentinelFile) {
// Used for stages the coordinator itself spawns: the very first stage of a
// task (planner), implementer retries and planner-revise (both are the
// coordinator's own branch decision after reading judge's verdict -- see
// runTaskInteractive), and phase-judge. Stages self-spawned by their
// predecessor (see spawnCommandLine and the instruction builders above) go
// through waitForStage below instead, since the coordinator never issued
// the spawn itself.
async function spawnAndWaitStage(cwd, task, stageLabel, stagePrompt, sentinelFile) {
const label = `${taskSessionLabel(task)}-${stageLabel}`;
fs.rmSync(sentinelFile, { force: true }); fs.rmSync(sentinelFile, { force: true });
const label = `${repoId}-${role}`;
let target = pool[role];
const spawnArgs = ["spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--prompt", stagePrompt]; if (!target) {
const { provider, model } = providerModelFor(stageLabel); const spawnArgs = ["spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--prompt", HARD_RULES + prompt];
const { provider, model } = providerModelFor(role);
if (provider) spawnArgs.push("--provider", provider); if (provider) spawnArgs.push("--provider", provider);
if (model) spawnArgs.push("--model", model); if (model) spawnArgs.push("--model", model);
const spawned = await runCmd(AGENT_MANAGER_BIN, spawnArgs); const spawned = await runCmd(AGENT_MANAGER_BIN, spawnArgs);
if (spawned.code !== 0) return { ok: false, crashed: true, error: spawned.out, sessionName: label }; if (spawned.code !== 0) return { ok: false, crashed: true, error: spawned.out, sessionName: label };
const sessionId = spawned.out; target = amSessionName(spawned.out);
const target = amSessionName(sessionId); pool[role] = target;
} else {
await runAmTmux(["send-keys", "-t", target, HARD_RULES + prompt, "Enter"]);
}
let ok = await waitForSentinel(sentinelFile, STAGE_TIMEOUT_MS); let ok = await waitForSentinel(sentinelFile, STAGE_TIMEOUT_MS);
if (!ok) { if (!ok) {
const pane = await runAmTmux(["capture-pane", "-t", target, "-p", "-S", "-200"]); const pane = await runAmTmux(["capture-pane", "-t", target, "-p", "-S", "-200"]);
const resolution = await askResolver( const resolution = await askResolver(
cwd, cwd,
task, repoId,
`Task ${task}'s "${stageLabel}" stage hasn't finished after 10 minutes. Its pane tail:\n${pane.out.slice(-3000)}\n\n` + `Repo ${repoId}'s "${role}" agent hasn't finished its current task after 10 minutes. Its pane tail:\n${pane.out.slice(-3000)}\n\n` +
`Decide: is it still making real progress and worth nudging to wrap up, or stuck and worth abandoning?` `Decide: is it still making real progress and worth nudging to wrap up, or stuck and worth abandoning?`
); );
if (resolution === "RETRY") { if (resolution === "RETRY") {
await runAmTmux([ await runAmTmux(["send-keys", "-t", target, `Please wrap up now and touch ${path.basename(sentinelFile)} when done.`, "Enter"]);
"send-keys",
"-t",
target,
`Please wrap up the "${stageLabel}" stage now and touch ${path.basename(sentinelFile)} when done.`,
"Enter",
]);
ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS);
}
}
return { ok, sessionName: label, sessionId };
}
// Waits for a stage that spawned itself (its predecessor's prompt embedded
// the agent-manager spawn command -- see spawnCommandLine). No session id
// is available here for pane capture on a timeout, since the coordinator
// never issued the spawn; the resolver gets a diagnostic prompt pointing at
// the session name to attach to directly instead, and RETRY just gives it
// the nudge window rather than an actual send-keys nudge.
async function waitForStage(cwd, task, stageLabel, sentinelFile) {
const label = `${taskSessionLabel(task)}-${stageLabel}`;
let ok = await waitForSentinel(sentinelFile, STAGE_TIMEOUT_MS);
if (!ok) {
const resolution = await askResolver(
cwd,
task,
`Task ${task}'s "${stageLabel}" stage hasn't finished after 10 minutes. It was self-spawned by its ` +
`predecessor agent -- attach to its agent-manager session ("${label}") directly to inspect progress; no ` +
`pane tail is available here. Decide: is it still making real progress and worth waiting longer, or stuck ` +
`and worth abandoning?`
);
if (resolution === "RETRY") {
ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS); ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS);
} }
} }
return { ok, sessionName: label }; return { ok, sessionName: label };
} }
function plannerPrompt(task, specHint) {
return (
`Use the planner skill to draft PLAN.md for task ${task}, reading its spec (${specHint}). Commit PLAN.md. ` +
`Then run: touch .stage-done-${task}-planner`
);
}
function investigatorPrompt(task) {
return (
`Use the investigator skill to confirm PLAN.md against real sources for task ${task}, append findings, commit. ` +
`Then run: touch .stage-done-${task}-investigator`
);
}
function implementerPrompt(task, attempt, feedbackHint) {
return (
`Use the implementer skill to implement what the current PLAN.md specifies for task ${task} (commit as you go). ` +
`${feedbackHint} Then run: touch .stage-done-${task}-implementer-${attempt}`
);
}
function judgePrompt(task, baseBranch, attempt) {
return (
`Use the judge skill to review the diff against ${baseBranch}...HEAD for task ${task}. Write your verdict to ` +
`.task-result-${task} as a single "VERDICT: PASS" or "VERDICT: FAIL" line plus one line of rationale, then ` +
`run: touch .stage-done-${task}-judge-${attempt}`
);
}
const MAX_IMPLEMENT_ATTEMPTS = 5; const MAX_IMPLEMENT_ATTEMPTS = 5;
const MAX_PLAN_REVISIONS = 3; const MAX_PLAN_REVISIONS = 3;
// Runs one task in its own git worktree (see runPhase): planner drafts // Runs one task against the repo's shared role pool: planner drafts
// PLAN.md, investigator confirms it, then implementer and judge go back and // PLAN.md, investigator confirms it, then implementer and judge go back and
// forth -- judge's FAIL rationale lands in .task-result-<task>, which the // forth -- judge's FAIL rationale lands in .task-result-<task>, which the
// next implementer attempt is told to read and address. After // next implementer attempt is told to read and address. After
// MAX_IMPLEMENT_ATTEMPTS straight fails, planner is brought back in to // MAX_IMPLEMENT_ATTEMPTS straight fails, the SAME planner agent is asked to
// judge whether the *plan* itself is wrong, not just the implementation; if // judge whether the plan itself is wrong (it still remembers drafting it);
// so it revises PLAN.md and the implementer gets a fresh attempt budget // if so it revises PLAN.md and the implementer gets a fresh attempt budget.
// against the new plan. MAX_PLAN_REVISIONS caps this from looping forever // MAX_PLAN_REVISIONS caps this from looping forever on a task that's
// on a task that's genuinely stuck. // genuinely stuck. All work happens directly in cwd (the repo's one shared
async function runTaskInteractive(cwd, baseBranch, task, pipelineSession, judgeOnly) { // clone, currently checked out to the phase branch) -- no worktree, since
// only one implementer/judge exist per repo and tasks run strictly one at a
// time (see runPhase).
async function runTaskOnPool(cwd, baseBranch, task, pool, repoId, pipelineSession, judgeOnly) {
const resultFile = path.join(cwd, `.task-result-${task}`); const resultFile = path.join(cwd, `.task-result-${task}`);
fs.rmSync(resultFile, { force: true }); fs.rmSync(resultFile, { force: true });
const specHint = `the file under tasks/ starting with "${task}-"`; const specHint = `the file under tasks/ starting with "${task}-"`;
// spawnRole: coordinator issues the spawn itself (planner's first call, const stage = async (role, prompt, sentinel, displayLabel) => {
// implementer retries, planner-revise, the judgeOnly quick check). const label = displayLabel || role;
// waitRole: the stage was self-spawned by its predecessor (see pipelineSession.activeTasks[task] = { stage: label, startedAt: new Date().toISOString() };
// spawnCommandLine/instruction builders above) -- the coordinator only
// waits. Both still run the commitPending backstop sweep afterward and
// update pipelineSession's progress display, unchanged from before.
const spawnRole = async (stageLabel, prompt, sentinel) => {
pipelineSession.activeTasks[task] = { stage: stageLabel, startedAt: new Date().toISOString() };
logProgress(pipelineSession); logProgress(pipelineSession);
const result = await spawnAndWaitStage(cwd, task, stageLabel, HARD_RULES + prompt, sentinel); const result = await runOnPool(pool, cwd, repoId, role, prompt, sentinel);
await commitPending(cwd, `task: ${task} (${stageLabel})`); await commitPending(cwd, `task: ${task} (${label})`);
return result;
};
const waitRole = async (stageLabel, sentinel) => {
pipelineSession.activeTasks[task] = { stage: stageLabel, startedAt: new Date().toISOString() };
logProgress(pipelineSession);
const result = await waitForStage(cwd, task, stageLabel, sentinel);
await commitPending(cwd, `task: ${task} (${stageLabel})`);
return result; return result;
}; };
@@ -411,7 +327,7 @@ data:
}; };
if (judgeOnly) { if (judgeOnly) {
const quick = await spawnRole( const quick = await stage(
"judge", "judge",
`Task ${task} may already be implemented on this branch -- check ` + `Task ${task} may already be implemented on this branch -- check ` +
`\`git log --oneline --grep '${task}'\` and the current code directly against its spec ` + `\`git log --oneline --grep '${task}'\` and the current code directly against its spec ` +
@@ -430,16 +346,10 @@ data:
} }
} }
// Happy path is self-chained: planner spawns investigator spawns let result = await stage("planner", plannerPrompt(task, specHint), path.join(cwd, `.stage-done-${task}-planner`));
// implementer (attempt 1) spawns judge, each as its own last action (see
// the instruction builders above). The coordinator spawns only planner
// here, then just waits for each stage's sentinel in turn -- same
// per-stage timeout/resolver/commit backstop as before, just not the one
// issuing the spawn for the self-chained stages.
let result = await spawnRole("planner", plannerInstructions(cwd, task, baseBranch, specHint), path.join(cwd, `.stage-done-${task}-planner`));
if (!result.ok) return abandon("planner", result); if (!result.ok) return abandon("planner", result);
result = await waitRole("investigator", path.join(cwd, `.stage-done-${task}-investigator`)); result = await stage("investigator", investigatorPrompt(task), path.join(cwd, `.stage-done-${task}-investigator`));
if (!result.ok) return abandon("investigator", result); if (!result.ok) return abandon("investigator", result);
let planRevisions = 0; let planRevisions = 0;
@@ -447,13 +357,6 @@ data:
let verdict = null; let verdict = null;
let resultText = ""; let resultText = "";
let justRevisedPlan = false; let justRevisedPlan = false;
// Attempt 1's implementer was already self-spawned by investigator above
// (its exact prompt was pre-baked into investigatorInstructions with
// attempt=1 and an empty feedbackHint, matching what attempt 1 always
// computes below) -- the coordinator only waits for it. Every retry
// after that is the coordinator's own branch decision on judge's
// verdict, so it spawns implementer directly instead.
let implementerSelfChained = true;
while (true) { while (true) {
implementAttempt++; implementAttempt++;
@@ -465,22 +368,14 @@ data:
: ""; : "";
justRevisedPlan = false; justRevisedPlan = false;
if (implementerSelfChained) { result = await stage(
result = await waitRole("implementer", path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`));
} else {
result = await spawnRole(
"implementer", "implementer",
implementerInstructions(cwd, task, baseBranch, implementAttempt, feedbackHint), implementerPrompt(task, implementAttempt, feedbackHint),
path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`) path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`)
); );
}
implementerSelfChained = false;
if (!result.ok) return abandon("implementer", result, implementAttempt); if (!result.ok) return abandon("implementer", result, implementAttempt);
// Judge is always self-spawned by implementer, coordinator- or result = await stage("judge", judgePrompt(task, baseBranch, implementAttempt), path.join(cwd, `.stage-done-${task}-judge-${implementAttempt}`));
// self-chained alike -- implementerInstructions embeds the same spawn
// command either way.
result = await waitRole("judge", path.join(cwd, `.stage-done-${task}-judge-${implementAttempt}`));
if (!result.ok) return abandon("judge", result, implementAttempt); if (!result.ok) return abandon("judge", result, implementAttempt);
resultText = fs.existsSync(resultFile) ? fs.readFileSync(resultFile, "utf8") : ""; resultText = fs.existsSync(resultFile) ? fs.readFileSync(resultFile, "utf8") : "";
@@ -490,15 +385,16 @@ data:
if (implementAttempt >= MAX_IMPLEMENT_ATTEMPTS) { if (implementAttempt >= MAX_IMPLEMENT_ATTEMPTS) {
if (planRevisions >= MAX_PLAN_REVISIONS) break; if (planRevisions >= MAX_PLAN_REVISIONS) break;
planRevisions++; planRevisions++;
result = await spawnRole( result = await stage(
"planner-revise", "planner",
`Implementer failed judge review ${MAX_IMPLEMENT_ATTEMPTS} times in a row for task ${task}. Read PLAN.md, ` + `Implementer failed judge review ${MAX_IMPLEMENT_ATTEMPTS} times in a row for task ${task}. Read PLAN.md, ` +
`the judge's feedback in .task-result-${task}, and the current diff against ${baseBranch}...HEAD. Decide ` + `the judge's feedback in .task-result-${task}, and the current diff against ${baseBranch}...HEAD. Decide ` +
`whether the plan's approach itself is wrong, not just the implementation -- if so, revise PLAN.md and ` + `whether the plan's approach itself is wrong, not just the implementation -- if so, revise PLAN.md and ` +
`commit. If you change the approach, also use the investigator skill to confirm the new approach against ` + `commit. If you change the approach, also use the investigator skill to confirm the new approach against ` +
`real sources before committing. If the plan is sound, note why in PLAN.md and leave it as-is. Then run: ` + `real sources before committing. If the plan is sound, note why in PLAN.md and leave it as-is. Then run: ` +
`touch .stage-done-${task}-planner-revise-${planRevisions}`, `touch .stage-done-${task}-planner-revise-${planRevisions}`,
path.join(cwd, `.stage-done-${task}-planner-revise-${planRevisions}`) path.join(cwd, `.stage-done-${task}-planner-revise-${planRevisions}`),
"planner-revise"
); );
if (!result.ok) return abandon("planner-revise", result, planRevisions); if (!result.ok) return abandon("planner-revise", result, planRevisions);
implementAttempt = 0; implementAttempt = 0;
@@ -515,73 +411,17 @@ data:
return { task, status: verdict === "PASS" ? "done" : "done-with-concerns", judgeRationale: resultText }; return { task, status: verdict === "PASS" ? "done" : "done-with-concerns", judgeRationale: resultText };
} }
// How many tasks' independent chains can be mid-flight at once within a // Runs every task in a phase (no declared dependency between them) strictly
// phase. This is the real buffer against a single-instance backend: a // one at a time against the repo's shared role pool -- only one implementer/
// lock around agent-manager spawn wouldn't work (spawn is fire-and-forget // judge/etc. exists per repo, so there is no per-task concurrency to have
// -- it creates the tmux session and returns immediately, the actual LLM // here anymore (see REPO_CONCURRENCY below for where concurrency now
// call happens later, asynchronously, inside that detached session, so a // lives). No worktrees: every task commits directly onto phaseBranch in the
// lock held only for the spawn call releases before the call it's // one shared cwd.
// supposed to guard even starts). Ollama itself queues concurrent async function runPhase(cwd, baseBranch, phaseTasks, pool, repoId, pipelineSession) {
// requests to one loaded model rather than erroring, but each one waits
// out the others -- PHASE_CONCURRENCY=1 makes that explicit: tasks run
// strictly one after another, one chain ever active against the backend
// at a time. Bump it back up once there's more than one model instance.
const PHASE_CONCURRENCY = Number(process.env.PHASE_CONCURRENCY) || 3;
// Runs one phase (a batch of tasks with no declared dependency on each
// other) with up to PHASE_CONCURRENCY tasks in flight at once. Each task
// gets its own git worktree off workBranch -- concurrent pi sessions writing
// into one shared working tree would corrupt the index; worktrees share the
// same object database but give each task an isolated checkout. After a
// task's session ends, its branch is merged back into workBranch and pushed,
// one merge at a time (git ref updates aren't safe to run concurrently even
// though the worktrees themselves are isolated).
async function runPhase(pipelineId, cwd, workBranch, phaseTasks, pipelineSession) {
const entries = phaseTasks.map((t) => (typeof t === "string" ? { id: t, judgeOnly: false } : t)); const entries = phaseTasks.map((t) => (typeof t === "string" ? { id: t, judgeOnly: false } : t));
const worktrees = {};
for (const entry of entries) { for (const entry of entries) {
const task = entry.id; const result = await runTaskOnPool(cwd, baseBranch, entry.id, pool, repoId, pipelineSession, entry.judgeOnly);
const wtDir = path.join(WORK_DIR, pipelineId, `wt-${task.replace(/[^a-zA-Z0-9]/g, "-")}`);
const taskBranch = `task/${task}`;
const add = await runGit(cwd, ["worktree", "add", "-b", taskBranch, wtDir, workBranch]);
if (add.code !== 0) {
pipelineSession.taskResults.push({ task, status: "worktree-crashed", error: add.out });
continue;
}
worktrees[task] = { wtDir, taskBranch };
}
const runnable = entries.filter((e) => worktrees[e.id]);
await runConcurrent(runnable, PHASE_CONCURRENCY, async (entry) => {
const { wtDir } = worktrees[entry.id];
const result = await runTaskInteractive(wtDir, workBranch, entry.id, pipelineSession, entry.judgeOnly);
pipelineSession.taskResults.push(result); pipelineSession.taskResults.push(result);
return result;
});
for (const entry of runnable) {
const task = entry.id;
const { wtDir, taskBranch } = worktrees[task];
const result = pipelineSession.taskResults.find((r) => r.task === task);
const merge = await runGit(cwd, ["merge", "--no-ff", taskBranch, "-m", `merge: ${task}`]);
if (merge.code !== 0) {
await runGit(cwd, ["merge", "--abort"]);
if (result) {
result.status = "merge-conflict";
result.mergeError = merge.out;
}
} else {
const push = await runGit(cwd, ["push", "-u", "origin", workBranch]);
if (result) {
result.pushed = push.code === 0;
if (!result.pushed) result.pushError = push.out;
}
}
await runGit(cwd, ["worktree", "remove", wtDir, "--force"]);
await runGit(cwd, ["branch", "-D", taskBranch]);
logProgress(pipelineSession); logProgress(pipelineSession);
} }
} }
@@ -625,31 +465,27 @@ data:
} }
function logProgress(pipelineSession) { function logProgress(pipelineSession) {
console.log(`[pipeline ${pipelineSession.id}] ${JSON.stringify(pipelineSession)}`); console.log(`[repo ${pipelineSession.id}] ${JSON.stringify(pipelineSession)}`);
} }
// Runs one repo's full pipeline: clone, then phases strictly sequentially.
// tasks: array of phases, each phase an array of task ids with no declared // tasks: array of phases, each phase an array of task ids with no declared
// dependency on each other (e.g. [["T0.1","T0.2"], ["T1.1","T1.2","T1.3"]]). // dependency on each other (e.g. [["T0.1","T0.2"], ["T1.1","T1.2","T1.3"]]).
// A flat array of ids is also accepted and treated as one single phase. // A flat array of ids is also accepted and treated as one single phase. If
// Phases run strictly sequentially; tasks within a phase run concurrently, // omitted, phases are discovered from the repo's own tasks/INDEX.md and run
// each in its own worktree -- see runPhase. Each phase gets its own branch // judgeOnly first (a cheap "is this already done" check against the
// (agent-run/<id>/<phaseLabel>, e.g. .../T1); once every task in that // board's possibly-stale checkmarks). Each phase gets its own branch
// phase lands "done" or "done-with-concerns" AND the phase judge passes // (agent-run/<repoId>/<phaseLabel>, e.g. .../T1); once every task in that
// the integration review, the phase branch is squash-merged into // phase lands "done" or "done-with-concerns" AND the phase judge (the same
// baseBranch and pushed, then the next phase branches off that updated // pooled judge agent that reviewed each task) passes the integration
// base. Any failure halts the pipeline before merging. // review, the phase branch is squash-merged into baseBranch and pushed,
async function runPipeline({ pipelineId, repo, baseBranch, tasks, branchName }) { // then the next phase branches off that updated base. Any failure halts
const cwd = path.join(WORK_DIR, pipelineId); // this repo's pipeline before merging -- it does not affect other repos
// running concurrently (see runCoordinator).
async function runRepoPipeline({ repoId, repo, baseBranch, tasks, branchName }, pipelineSession) {
const cwd = path.join(WORK_DIR, repoId);
fs.mkdirSync(cwd, { recursive: true }); fs.mkdirSync(cwd, { recursive: true });
const pool = {};
const pipelineSession = {
id: pipelineId,
status: "running",
taskResults: [],
activeTasks: {},
totalTasks: 0,
startedAt: new Date().toISOString(),
};
const finish = (status) => { const finish = (status) => {
pipelineSession.status = status; pipelineSession.status = status;
@@ -670,15 +506,9 @@ data:
} }
if (!fs.existsSync(path.join(cwd, ".git"))) return finish("clone-missing"); if (!fs.existsSync(path.join(cwd, ".git"))) return finish("clone-missing");
// No --tasks given: discover the phase/task board from the repo's own
// tasks/INDEX.md (see parseTaskBoard) instead of requiring the caller to
// already know it. Each discovered task runs judgeOnly first -- a cheap
// "is this already done on this branch" check against the board's own
// possibly-stale checkmarks, falling through to the full planner/
// investigator/implementer/judge flow only when it isn't.
let phases = tasks ? (Array.isArray(tasks[0]) ? tasks : [tasks]) : parseTaskBoard(cwd); let phases = tasks ? (Array.isArray(tasks[0]) ? tasks : [tasks]) : parseTaskBoard(cwd);
if (!phases || phases.length === 0) { if (!phases || phases.length === 0) {
pipelineSession.gitError = "no --tasks given and tasks/INDEX.md not found or empty"; pipelineSession.gitError = "no tasks given and tasks/INDEX.md not found or empty";
return finish("no-tasks-found"); return finish("no-tasks-found");
} }
if (!tasks) { if (!tasks) {
@@ -689,7 +519,7 @@ data:
for (let i = 0; i < phases.length; i++) { for (let i = 0; i < phases.length; i++) {
const phaseTasks = phases[i]; const phaseTasks = phases[i];
const phaseLabel = phaseLabelFor(phaseTasks, i); const phaseLabel = phaseLabelFor(phaseTasks, i);
const phaseBranch = branchName ? `${branchName}/${phaseLabel}` : `agent-run/${pipelineId}/${phaseLabel}`; const phaseBranch = branchName ? `${branchName}/${phaseLabel}` : `agent-run/${repoId}/${phaseLabel}`;
const branchResult = await runGit(cwd, ["checkout", "-b", phaseBranch]); const branchResult = await runGit(cwd, ["checkout", "-b", phaseBranch]);
if (branchResult.code !== 0) { if (branchResult.code !== 0) {
@@ -720,7 +550,7 @@ data:
await runGit(cwd, ["commit", "-m", "chore: broaden .gitignore for agent-run artifacts"]); await runGit(cwd, ["commit", "-m", "chore: broaden .gitignore for agent-run artifacts"]);
} }
await runPhase(pipelineId, cwd, phaseBranch, phaseTasks, pipelineSession); await runPhase(cwd, phaseBranch, phaseTasks, pool, repoId, pipelineSession);
const phaseTaskIds = new Set(phaseTasks.map((t) => (typeof t === "string" ? t : t.id))); const phaseTaskIds = new Set(phaseTasks.map((t) => (typeof t === "string" ? t : t.id)));
const phaseResults = pipelineSession.taskResults.filter((r) => phaseTaskIds.has(r.task)); const phaseResults = pipelineSession.taskResults.filter((r) => phaseTaskIds.has(r.task));
@@ -734,12 +564,11 @@ data:
const phaseResultFile = path.join(cwd, `.phase-result-${phaseLabel}`); const phaseResultFile = path.join(cwd, `.phase-result-${phaseLabel}`);
fs.rmSync(phaseResultFile, { force: true }); fs.rmSync(phaseResultFile, { force: true });
const phaseSentinel = path.join(cwd, `.stage-done-phase-${phaseLabel}-judge`); const phaseJudge = await runOnPool(
const phaseJudge = await spawnAndWaitStage( pool,
cwd, cwd,
`phase-${phaseLabel}`, repoId,
"judge", "judge",
HARD_RULES +
`Use the judge skill to review the full phase diff for phase ${phaseLabel} against ` + `Use the judge skill to review the full phase diff for phase ${phaseLabel} against ` +
`${baseBranch}...HEAD (covers every task in this phase: ${[...phaseTaskIds].join(", ")}). Every ` + `${baseBranch}...HEAD (covers every task in this phase: ${[...phaseTaskIds].join(", ")}). Every ` +
`individual task already passed its own judge review -- your job here is different: confirm the ` + `individual task already passed its own judge review -- your job here is different: confirm the ` +
@@ -747,7 +576,7 @@ data:
`each task's isolated unit checks) exist and actually exercise the phase's intended use case end ` + `each task's isolated unit checks) exist and actually exercise the phase's intended use case end ` +
`to end. Write your verdict to .phase-result-${phaseLabel} as a single "VERDICT: PASS" or ` + `to end. Write your verdict to .phase-result-${phaseLabel} as a single "VERDICT: PASS" or ` +
`"VERDICT: FAIL" line plus rationale, then run: touch .stage-done-phase-${phaseLabel}-judge`, `"VERDICT: FAIL" line plus rationale, then run: touch .stage-done-phase-${phaseLabel}-judge`,
phaseSentinel path.join(cwd, `.stage-done-phase-${phaseLabel}-judge`)
); );
await commitPending(cwd, `phase: ${phaseLabel} integration review`); await commitPending(cwd, `phase: ${phaseLabel} integration review`);
@@ -791,11 +620,57 @@ data:
return finish("completed"); return finish("completed");
} }
async function runConcurrent(items, limit, worker) {
const results = new Array(items.length);
let i = 0;
async function next() {
while (i < items.length) {
const idx = i++;
results[idx] = await worker(items[idx], idx);
}
}
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, next));
return results;
}
// How many repos can be mid-flight at once. Each repo gets its own clone
// and its own 4-agent pool (planner/investigator/implementer/judge), so
// this is now the real concurrency knob -- tasks within one repo are
// already serialized against that repo's pool (see runPhase). Backends
// that queue rather than reject concurrent requests to one loaded model
// (e.g. Ollama) still process this many repos' worth of stages one at a
// time internally even if REPO_CONCURRENCY says otherwise; bump it once
// there's more than one model instance to actually parallelize against.
const REPO_CONCURRENCY = Number(process.env.REPO_CONCURRENCY) || 3;
// Top-level entry point: runs every repo in `repos` to completion, up to
// REPO_CONCURRENCY at a time. Returns a map of repoId -> final
// pipelineSession, one per repo, independent of how the others fared.
async function runCoordinator({ repos, base, tasks, branchName }) {
const sessions = {};
await runConcurrent(repos, REPO_CONCURRENCY, async (repoUrl) => {
const repoId = crypto.randomUUID();
const pipelineSession = {
id: repoId,
repo: repoUrl,
status: "running",
taskResults: [],
activeTasks: {},
totalTasks: 0,
startedAt: new Date().toISOString(),
};
sessions[repoId] = pipelineSession;
await runRepoPipeline({ repoId, repo: repoUrl, baseBranch: base, tasks, branchName }, pipelineSession);
});
return sessions;
}
function parseArgs(argv) { function parseArgs(argv) {
const opts = { base: "main" }; const opts = { base: "main" };
for (let i = 0; i < argv.length; i++) { for (let i = 0; i < argv.length; i++) {
const a = argv[i]; const a = argv[i];
if (a === "--repo") opts.repo = argv[++i]; if (a === "--repo") opts.repo = argv[++i];
else if (a === "--repos") opts.repos = argv[++i];
else if (a === "--base") opts.base = argv[++i]; else if (a === "--base") opts.base = argv[++i];
else if (a === "--tasks") opts.tasks = argv[++i]; else if (a === "--tasks") opts.tasks = argv[++i];
else if (a === "--branch") opts.branch = argv[++i]; else if (a === "--branch") opts.branch = argv[++i];
@@ -805,10 +680,13 @@ data:
async function main() { async function main() {
const opts = parseArgs(process.argv.slice(2)); const opts = parseArgs(process.argv.slice(2));
if (!opts.repo) { const repos = opts.repos ? opts.repos.split(",") : opts.repo ? [opts.repo] : null;
if (!repos || repos.length === 0) {
console.error( console.error(
"usage: coordinator.js --repo <url> [--tasks T0.1,T0.2;T1.1,T1.2,...] [--base main] [--branch <name>]\n" + "usage: coordinator.js --repos <url1,url2,...> [--tasks T0.1,T0.2;T1.1,T1.2,...] [--base main] [--branch <name>]\n" +
" --tasks omitted: discovers phases from the cloned repo's own tasks/INDEX.md" " --repo <url> also accepted for a single repo\n" +
" --tasks applies to every repo listed; omitted: each repo discovers its own phases from tasks/INDEX.md\n" +
" REPO_CONCURRENCY env var (default 3): how many repos run at once"
); );
// process.exitCode + natural exit, not process.exit() -- stdout piped // process.exitCode + natural exit, not process.exit() -- stdout piped
// through kubectl exec (not a TTY) can drop buffered console.log/ // through kubectl exec (not a TTY) can drop buffered console.log/
@@ -819,13 +697,12 @@ data:
return; return;
} }
const phases = opts.tasks ? opts.tasks.split(";").map((phase) => phase.split(",")) : null; const phases = opts.tasks ? opts.tasks.split(";").map((phase) => phase.split(",")) : null;
const pipelineId = require("node:crypto").randomUUID(); const sessions = await runCoordinator({ repos, base: opts.base, tasks: phases, branchName: opts.branch });
const result = await runPipeline({ pipelineId, repo: opts.repo, baseBranch: opts.base, tasks: phases, branchName: opts.branch }); process.exitCode = Object.values(sessions).every((s) => s.status === "completed") ? 0 : 1;
process.exitCode = result.status === "completed" ? 0 : 1;
} }
if (require.main === module) { if (require.main === module) {
main(); main();
} }
module.exports = { runPipeline, spawnCommandLine, spawnAndWaitStage, waitForStage, parseTaskBoard }; module.exports = { runCoordinator, runRepoPipeline, runOnPool, parseTaskBoard };
k8s/apps/agent-pod/deployment.yaml
+11 -9
View File
@@ -31,15 +31,17 @@ spec:
# Also builds the agent-manager fork (github.com/Riotpiaole/ # Also builds the agent-manager fork (github.com/Riotpiaole/
# agent-manager, add-headless-spawn branch) from source and drops # agent-manager, add-headless-spawn branch) from source and drops
# coordinator.js in beside hub.js -- neither is the container's # coordinator.js in beside hub.js -- neither is the container's
# foreground process. hub.js keeps that role unchanged; running # foreground process. hub.js keeps that role unchanged; coordinator.js
# multiple repos' pipelines concurrently in this one pod means # itself now owns multi-repo concurrency (REPO_CONCURRENCY env,
# `kubectl exec <pod> -- node /root/coordinator.js --repo X --tasks # default 3), so one invocation handles every repo:
# Y &` once per repo, each an independent process inside the same # `kubectl exec <pod> -- node /root/coordinator.js --repos
# container, each spawning its own agent-manager-tracked sessions # repoA,repoB,... --tasks ...`. Each repo gets its own clone and its
# on the container's local tmux server -- `kubectl exec -it <pod> # own persistent 4-agent pool (planner/investigator/implementer/
# -- agent-manager` attaches its TUI live against those same # judge, one agent-manager session per role, reused across every
# sessions, no cross-machine visibility problem since spawner, # task in that repo) on the container's local tmux server --
# tmux server, and viewer are all colocated here. # `kubectl exec -it <pod> -- agent-manager` attaches its TUI live
# against those same sessions, no cross-machine visibility problem
# since spawner, tmux server, and viewer are all colocated here.
# #
# No prebuilt Linux binary is shipped for agent-manager: the local # No prebuilt Linux binary is shipped for agent-manager: the local
# .bin/ build is macOS arm64 (wrong OS/arch for this container # .bin/ build is macOS arm64 (wrong OS/arch for this container