781 lines
38 KiB
YAML
781 lines
38 KiB
YAML
apiVersion: v1
|
|
kind: ConfigMap
|
|
metadata:
|
|
name: coordinator-src
|
|
namespace: agent-pod
|
|
data:
|
|
coordinator.js: |
|
|
#!/usr/bin/env node
|
|
// coordinator: local CLI that drives a multi-phase, multi-task pipeline of
|
|
// planner/investigator/implementer/judge stages, same state machine as
|
|
// hub.js's old /pipeline handler. Every interactive stage runs through the
|
|
// patched agent-manager fork's `spawn` subcommand, so it is a real tracked
|
|
// session (tmux pane on agent-manager's private socket + a state.db row)
|
|
// from the moment it exists -- attachable and visible in agent-manager's
|
|
// own TUI the whole time it runs.
|
|
//
|
|
// Per repo, each role (planner/investigator/implementer/judge) is ONE
|
|
// persistent agent-manager session, not a fresh spawn per task: the first
|
|
// task to need a role spawns it, every later task for that role reuses the
|
|
// same tmux pane via `tmux send-keys` (see runOnPool) -- the same nudge
|
|
// mechanism that used to only fire on a stall now doubles as "give this
|
|
// agent its next task." Every role reads everything it needs fresh off disk
|
|
// each call, so every pane gets a `/new` before every reuse instead of
|
|
// accumulating history that degrades and eventually errors out task after
|
|
// task -- same pane, same agent-manager session, zero memory of the last
|
|
// task it handled.
|
|
// 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.
|
|
const fs = require("node:fs");
|
|
const path = require("node:path");
|
|
const { spawn } = require("node:child_process");
|
|
|
|
const WORK_DIR = process.env.HUB_WORK_DIR || path.join(require("node:os").tmpdir(), "agent-harness-work");
|
|
|
|
// Never rely on a bare `pi`/`agent-manager` on $PATH -- see PI_BIN's own
|
|
// comment below; the same collision risk applies to any CLI name. Always
|
|
// invoke explicit pinned paths.
|
|
const PI_BIN =
|
|
process.env.PI_BIN ||
|
|
path.join(__dirname, "..", ".pi-cli", "node_modules", "@earendil-works", "pi-coding-agent", "dist", "cli.js");
|
|
|
|
const AGENT_MANAGER_BIN = process.env.AGENT_MANAGER_BIN || path.join(__dirname, "..", ".bin", "agent-manager-fork");
|
|
|
|
// 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
|
|
// route every stage -- headless (spawnPi) and interactive (runOnPool) --
|
|
// at the homelab model instead, e.g. AGENT_PROVIDER=homelab-ornith
|
|
// AGENT_MODEL=ornith:35b.
|
|
const AGENT_PROVIDER = process.env.AGENT_PROVIDER || "";
|
|
const AGENT_MODEL = process.env.AGENT_MODEL || "";
|
|
|
|
// judge can run a different model than the rest of the chain, e.g.
|
|
// homelab-reasoning instead of homelab-ornith now that verifier/PRM is
|
|
// retired. Falls back to AGENT_PROVIDER/AGENT_MODEL when unset, so a run
|
|
// that doesn't care keeps one uniform model everywhere.
|
|
const JUDGE_PROVIDER = process.env.JUDGE_PROVIDER || AGENT_PROVIDER;
|
|
const JUDGE_MODEL = process.env.JUDGE_MODEL || AGENT_MODEL;
|
|
|
|
function providerModelFor(role) {
|
|
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
|
|
// (internal/tmux/tmux.go: defaultSocket = "agentmgr", sessionName(id) =
|
|
// "am_"+id) -- stable, documented internals of the fork, used here only
|
|
// for read-only introspection (pane capture) and role nudges, exactly the
|
|
// class of operation hub.js already ran directly against its own sessions
|
|
// rather than asking a model to do it.
|
|
const AM_SOCKET = "agentmgr";
|
|
function amSessionName(id) {
|
|
return `am_${id}`;
|
|
}
|
|
function runAmTmux(args) {
|
|
return runCmd("tmux", ["-L", AM_SOCKET, ...args]);
|
|
}
|
|
|
|
const ROLE_SKILLS = new Set(["planner", "investigator", "info-collector", "implementer", "judge", "resolver"]);
|
|
|
|
// Every role reads everything it needs fresh off disk each call -- PLAN.md,
|
|
// the task spec, judge's verdict file, `git diff` against baseBranch --
|
|
// nothing depends on remembering earlier tasks. Left to accumulate, a
|
|
// pooled session's conversation grows without bound across every task in a
|
|
// repo and both correctness and reliability degrade hard once it does
|
|
// (observed: a planner session at ~1.5M cumulative tokens started erroring
|
|
// out every call, an investigator session that far gone started narrating a
|
|
// different codebase entirely). So every role gets reset to a clean
|
|
// conversation before every reuse instead of just being nudged with the
|
|
// next prompt -- same pane, same agent-manager session (still
|
|
// visible/attachable), zero history carried between tasks.
|
|
|
|
function sleep(ms) {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
const HARD_RULES =
|
|
"Read and follow ~/.pi/agent/skills/karpathy-guidelines/SKILL.md and " +
|
|
"~/.pi/agent/skills/caveman/SKILL.md as hard rules for this entire task, before anything else. ";
|
|
|
|
function parseVerdictLine(text, label) {
|
|
if (!text) return null;
|
|
const re = new RegExp(`${label}:\\s*(\\w+)`, "i");
|
|
const m = text.match(re);
|
|
return m ? m[1].toUpperCase() : null;
|
|
}
|
|
|
|
function runCmd(bin, args, cwd) {
|
|
return new Promise((resolve) => {
|
|
const child = spawn(bin, args, { cwd, stdio: ["ignore", "pipe", "pipe"] });
|
|
let out = "";
|
|
child.stdout.on("data", (c) => (out += c));
|
|
child.stderr.on("data", (c) => (out += c));
|
|
child.on("close", (code) => resolve({ code, out: out.trim() }));
|
|
});
|
|
}
|
|
|
|
function runGit(cwd, args) {
|
|
return runCmd("git", args, cwd);
|
|
}
|
|
|
|
// A stage saying "commit" in its prompt is a request, not a guarantee -- seen
|
|
// in practice: a stage writes a real file and simply never runs `git add`/
|
|
// `git commit`, leaving it untracked and invisible to every later `git diff`.
|
|
// Sweep and commit anything left dirty after every stage, deterministically.
|
|
async function commitPending(cwd, message) {
|
|
await runGit(cwd, ["add", "-A"]);
|
|
const status = await runGit(cwd, ["status", "--porcelain"]);
|
|
if (!status.out) return { committed: false };
|
|
const commit = await runGit(cwd, ["commit", "-m", message]);
|
|
return { committed: commit.code === 0, error: commit.code !== 0 ? commit.out : undefined };
|
|
}
|
|
|
|
// 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
|
|
// session: resolver's crash/stall diagnosis, and the initial clone. Kept
|
|
// exactly as before -- only the interactive per-role stages (runOnPool,
|
|
// below) go through agent-manager.
|
|
function spawnPi({ agent, prompt, cwd }) {
|
|
const finalPrompt = ROLE_SKILLS.has(agent) ? `/skill:${agent} ${HARD_RULES}${prompt}` : prompt;
|
|
const args = ["-p", "--mode", "json"];
|
|
if (AGENT_PROVIDER) args.push("--provider", AGENT_PROVIDER);
|
|
if (AGENT_MODEL) args.push("--model", AGENT_MODEL);
|
|
args.push(finalPrompt);
|
|
const child = spawn(PI_BIN, args, { stdio: ["ignore", "pipe", "pipe"], cwd });
|
|
let lastText = "";
|
|
let stderrTail = "";
|
|
let buf = "";
|
|
child.stdout.on("data", (chunk) => {
|
|
buf += chunk;
|
|
let idx;
|
|
while ((idx = buf.indexOf("\n")) !== -1) {
|
|
const line = buf.slice(0, idx);
|
|
buf = buf.slice(idx + 1);
|
|
if (!line.trim()) continue;
|
|
try {
|
|
const event = JSON.parse(line);
|
|
if (event.type === "message_end" && event.message && Array.isArray(event.message.content)) {
|
|
const text = event.message.content
|
|
.filter((c) => c.type === "text")
|
|
.map((c) => c.text)
|
|
.join("\n");
|
|
if (text) lastText = text;
|
|
}
|
|
} catch {
|
|
// non-JSON stdout noise, ignore
|
|
}
|
|
}
|
|
});
|
|
child.stderr.on("data", (chunk) => {
|
|
process.stderr.write(chunk);
|
|
stderrTail = (stderrTail + chunk.toString()).slice(-4000);
|
|
});
|
|
return new Promise((resolve) => {
|
|
child.on("close", (code) => resolve({ code, lastText, stderrTail }));
|
|
});
|
|
}
|
|
|
|
async function askResolver(cwd, repoId, diagnosticPrompt) {
|
|
const result = await spawnPi({ agent: "resolver", prompt: diagnosticPrompt, cwd });
|
|
return parseVerdictLine(result.lastText, "RESOLUTION");
|
|
}
|
|
|
|
const STAGE_TIMEOUT_MS = 10 * 60 * 1000;
|
|
const NUDGE_TIMEOUT_MS = 5 * 60 * 1000;
|
|
|
|
// Resolves as soon as filePath appears (fs.watch on its directory), or
|
|
// after limitMs with no sign of it -- event-driven completion instead of
|
|
// hub.js's old 10s poll.
|
|
function waitForSentinel(filePath, limitMs) {
|
|
return new Promise((resolve) => {
|
|
if (fs.existsSync(filePath)) return resolve(true);
|
|
const dir = path.dirname(filePath);
|
|
let settled = false;
|
|
let watcher;
|
|
const finish = (result) => {
|
|
if (settled) return;
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
if (watcher) {
|
|
try {
|
|
watcher.close();
|
|
} catch {
|
|
// already closed
|
|
}
|
|
}
|
|
resolve(result);
|
|
};
|
|
try {
|
|
watcher = fs.watch(dir, () => {
|
|
if (fs.existsSync(filePath)) finish(true);
|
|
});
|
|
} catch {
|
|
// dir missing at watch time is a real bug elsewhere (cwd should
|
|
// already exist); surface it as a timeout rather than hang forever.
|
|
return finish(false);
|
|
}
|
|
// Closes the race between the existsSync check above and the watcher
|
|
// actually being attached.
|
|
if (fs.existsSync(filePath)) return finish(true);
|
|
const timer = setTimeout(() => finish(false), limitMs);
|
|
});
|
|
}
|
|
|
|
// Runs one task's worth of work on a persistent per-role agent: spawns the
|
|
// role's session the first time it's ever needed for this repo, sends every
|
|
// later prompt into that same tmux pane via send-keys -- prefixed with a
|
|
// `/new` first, so the pane and agent-manager session stay the same but the
|
|
// model starts that prompt with a clean conversation, no history carried
|
|
// over from whatever task this role last handled. pool is a plain object
|
|
// keyed by role name ("planner"/"investigator"/"implementer"/"judge"),
|
|
// shared across every task in a repo's pipeline (see runRepoPipeline) -- it
|
|
// IS the 4-agent pool, one entry per role, filled in lazily as each role
|
|
// gets its first task.
|
|
async function runOnPool(pool, cwd, repoId, role, prompt, sentinelFile) {
|
|
fs.rmSync(sentinelFile, { force: true });
|
|
const label = `${repoId}-${role}`;
|
|
let target = pool[role];
|
|
|
|
if (!target) {
|
|
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 (model) spawnArgs.push("--model", model);
|
|
const spawned = await runCmd(AGENT_MANAGER_BIN, spawnArgs);
|
|
if (spawned.code !== 0) return { ok: false, crashed: true, error: spawned.out, sessionName: label };
|
|
target = amSessionName(spawned.out);
|
|
pool[role] = target;
|
|
} else {
|
|
await runAmTmux(["send-keys", "-t", target, "/new", "Enter"]);
|
|
await sleep(1000);
|
|
await runAmTmux(["send-keys", "-t", target, HARD_RULES + prompt, "Enter"]);
|
|
}
|
|
|
|
let ok = await waitForSentinel(sentinelFile, STAGE_TIMEOUT_MS);
|
|
if (!ok) {
|
|
const pane = await runAmTmux(["capture-pane", "-t", target, "-p", "-S", "-200"]);
|
|
const resolution = await askResolver(
|
|
cwd,
|
|
repoId,
|
|
`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?`
|
|
);
|
|
if (resolution === "RETRY") {
|
|
await runAmTmux(["send-keys", "-t", target, `Please wrap up now and touch ${path.basename(sentinelFile)} when done.`, "Enter"]);
|
|
ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS);
|
|
}
|
|
}
|
|
return { ok, sessionName: label };
|
|
}
|
|
|
|
function plannerPrompt(task, specHint, judgeOnly) {
|
|
// judgeOnly (auto-discovered tasks only, see parseTaskBoard): planner
|
|
// itself decides whether the task is already done before planning it,
|
|
// reading tasks/INDEX.md's own status notes plus git log/current code --
|
|
// replaces what used to be a separate judge pre-check call. One LLM round
|
|
// trip instead of two, and the same agent that's about to plan the task
|
|
// is the one deciding whether planning it is even necessary.
|
|
const decideStep = judgeOnly
|
|
? `First, decide whether task ${task} is already fully implemented on this branch: check ` +
|
|
`\`git log --oneline --grep '${task}'\`, tasks/INDEX.md's own status notes for this task, and the current ` +
|
|
`code directly against its spec (${specHint})'s acceptance criteria. Write your decision to ` +
|
|
`.task-result-${task} as a single "VERDICT: PASS" (already done, no further work needed) or ` +
|
|
`"VERDICT: FAIL" (needs work) line plus one line of rationale. If VERDICT is FAIL, continue below and ` +
|
|
`draft the plan in this same turn; if VERDICT is PASS, skip the rest and go straight to the touch step.\n\n`
|
|
: "";
|
|
return (
|
|
`${decideStep}Use the planner skill to draft PLAN.md for task ${task}, reading its spec (${specHint}). ` +
|
|
`PLAN.md is scratch state for this harness, not a deliverable -- do NOT commit it or add it to git. ` +
|
|
`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. ` +
|
|
`PLAN.md is scratch state for this harness, not a deliverable -- do NOT commit it or add it to git. ` +
|
|
`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_PLAN_REVISIONS = 3;
|
|
|
|
// Runs one task against the repo's shared role pool: planner drafts
|
|
// PLAN.md (for auto-discovered tasks, first deciding off tasks/INDEX.md and
|
|
// the repo's own state whether the task is already done -- see
|
|
// plannerPrompt's judgeOnly branch; judge never does this pre-check),
|
|
// investigator confirms it, then implementer and judge go back and
|
|
// forth -- judge's FAIL rationale lands in .task-result-<task>, which the
|
|
// next implementer attempt is told to read and address. After
|
|
// MAX_IMPLEMENT_ATTEMPTS straight fails, the planner role is asked to judge
|
|
// whether the plan itself is wrong -- fresh conversation, same as any other
|
|
// planner call, reading PLAN.md/the judge feedback/the
|
|
// diff off disk rather than remembering having drafted the original plan.
|
|
// If it decides the approach is wrong it revises PLAN.md and the implementer
|
|
// gets a fresh attempt budget.
|
|
// MAX_PLAN_REVISIONS caps this from looping forever on a task that's
|
|
// genuinely stuck. All work happens directly in cwd (the repo's one shared
|
|
// 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}`);
|
|
fs.rmSync(resultFile, { force: true });
|
|
|
|
const specHint = `the file under tasks/ starting with "${task}-"`;
|
|
|
|
const stage = async (role, prompt, sentinel, displayLabel) => {
|
|
const label = displayLabel || role;
|
|
pipelineSession.activeTasks[task] = { stage: label, startedAt: new Date().toISOString() };
|
|
logProgress(pipelineSession);
|
|
const result = await runOnPool(pool, cwd, repoId, role, prompt, sentinel);
|
|
await commitPending(cwd, `task: ${task} (${label})`);
|
|
return result;
|
|
};
|
|
|
|
const abandon = (stageLabel, result, attempt) => {
|
|
delete pipelineSession.activeTasks[task];
|
|
logProgress(pipelineSession);
|
|
return {
|
|
task,
|
|
status: result.crashed ? "spawn-crashed" : "timed-out",
|
|
error: result.error,
|
|
stoppedAt: stageLabel,
|
|
...(attempt !== undefined ? { attempt } : {}),
|
|
};
|
|
};
|
|
|
|
// PLAN.md is scratch state for this one task, not a deliverable (see
|
|
// plannerPrompt/investigatorPrompt -- it's gitignored too, as a backstop
|
|
// in case an agent commits it anyway). Discard it once the task is done,
|
|
// whatever the outcome, so it never bleeds into the next task's planner
|
|
// call or sits around as stale harness clutter in the shared clone.
|
|
try {
|
|
let result = await stage("planner", plannerPrompt(task, specHint, judgeOnly), path.join(cwd, `.stage-done-${task}-planner`));
|
|
if (!result.ok) return abandon("planner", result);
|
|
|
|
if (judgeOnly) {
|
|
const quickText = fs.existsSync(resultFile) ? fs.readFileSync(resultFile, "utf8") : "";
|
|
if (parseVerdictLine(quickText, "VERDICT") === "PASS") {
|
|
delete pipelineSession.activeTasks[task];
|
|
logProgress(pipelineSession);
|
|
return { task, status: "done", judgeRationale: quickText, judgeOnlyPass: true };
|
|
}
|
|
}
|
|
|
|
result = await stage("investigator", investigatorPrompt(task), path.join(cwd, `.stage-done-${task}-investigator`));
|
|
if (!result.ok) return abandon("investigator", result);
|
|
|
|
let planRevisions = 0;
|
|
let implementAttempt = 0;
|
|
let verdict = null;
|
|
let resultText = "";
|
|
let justRevisedPlan = false;
|
|
|
|
while (true) {
|
|
implementAttempt++;
|
|
const feedbackHint = fs.existsSync(resultFile)
|
|
? justRevisedPlan
|
|
? `.task-result-${task} holds the judge's feedback against the OLD plan, which prompted a plan revision -- ` +
|
|
`PLAN.md has since changed. Read the current PLAN.md as the source of truth, not the old feedback verbatim.`
|
|
: `A previous judge review exists at .task-result-${task} -- read it and address every issue it raises.`
|
|
: "";
|
|
justRevisedPlan = false;
|
|
|
|
result = await stage(
|
|
"implementer",
|
|
implementerPrompt(task, implementAttempt, feedbackHint),
|
|
path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`)
|
|
);
|
|
if (!result.ok) return abandon("implementer", result, implementAttempt);
|
|
|
|
result = await stage("judge", judgePrompt(task, baseBranch, implementAttempt), path.join(cwd, `.stage-done-${task}-judge-${implementAttempt}`));
|
|
if (!result.ok) return abandon("judge", result, implementAttempt);
|
|
|
|
resultText = fs.existsSync(resultFile) ? fs.readFileSync(resultFile, "utf8") : "";
|
|
verdict = parseVerdictLine(resultText, "VERDICT");
|
|
if (verdict === "PASS") break;
|
|
|
|
if (implementAttempt >= MAX_IMPLEMENT_ATTEMPTS) {
|
|
if (planRevisions >= MAX_PLAN_REVISIONS) break;
|
|
planRevisions++;
|
|
result = await stage(
|
|
"planner",
|
|
`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 ` +
|
|
`whether the plan's approach itself is wrong, not just the implementation -- if so, revise PLAN.md. If ` +
|
|
`you change the approach, also use the investigator skill to confirm the new approach against real ` +
|
|
`sources. If the plan is sound, note why in PLAN.md and leave it as-is. PLAN.md is scratch state for ` +
|
|
`this harness, not a deliverable -- do NOT commit it or add it to git. Then run: ` +
|
|
`touch .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);
|
|
implementAttempt = 0;
|
|
justRevisedPlan = true;
|
|
}
|
|
}
|
|
|
|
delete pipelineSession.activeTasks[task];
|
|
logProgress(pipelineSession);
|
|
|
|
if (verdict !== "PASS" && planRevisions >= MAX_PLAN_REVISIONS) {
|
|
return { task, status: "unresolved", judgeRationale: resultText, implementAttempts: implementAttempt, planRevisions };
|
|
}
|
|
return { task, status: verdict === "PASS" ? "done" : "done-with-concerns", judgeRationale: resultText };
|
|
} finally {
|
|
fs.rmSync(path.join(cwd, "PLAN.md"), { force: true });
|
|
}
|
|
}
|
|
|
|
// Runs every task in a phase (no declared dependency between them) strictly
|
|
// one at a time against the repo's shared role pool -- only one implementer/
|
|
// judge/etc. exists per repo, so there is no per-task concurrency to have
|
|
// here anymore (see REPO_CONCURRENCY below for where concurrency now
|
|
// lives). No worktrees: every task commits directly onto phaseBranch in the
|
|
// one shared cwd.
|
|
async function runPhase(cwd, baseBranch, phaseTasks, pool, repoId, pipelineSession) {
|
|
const entries = phaseTasks.map((t) => (typeof t === "string" ? { id: t, judgeOnly: false } : t));
|
|
for (const entry of entries) {
|
|
const result = await runTaskOnPool(cwd, baseBranch, entry.id, pool, repoId, pipelineSession, entry.judgeOnly);
|
|
pipelineSession.taskResults.push(result);
|
|
logProgress(pipelineSession);
|
|
}
|
|
}
|
|
|
|
// Discovers phases/tasks from the repo's own tasks/INDEX.md instead of
|
|
// requiring the caller to pass --tasks. Matches this convention's board
|
|
// shape (see e.g. Poimen/agent-rust's tasks/INDEX.md): a numbered phase
|
|
// heading ("## 1 — Foundations · T0.x"), followed by a markdown table
|
|
// whose rows link to each task's own spec file ("| [T0.1](T0.1-....md) |
|
|
// ... |"). Headings that aren't a numbered phase (prose sections like
|
|
// "## Ordering — declared, never derived", "## Progress") are skipped --
|
|
// only "## <digits> — ..." starts a new phase. Returns null if
|
|
// tasks/INDEX.md doesn't exist; an empty array if it exists but no phase
|
|
// yielded any task rows.
|
|
function parseTaskBoard(cwd) {
|
|
const indexPath = path.join(cwd, "tasks", "INDEX.md");
|
|
if (!fs.existsSync(indexPath)) return null;
|
|
|
|
const phaseHeaderRe = /^##\s+\d+\s+—/;
|
|
const taskRowRe = /^\|\s*\[([A-Za-z0-9.]+)\]\(/;
|
|
|
|
const phases = [];
|
|
let current = null;
|
|
for (const line of fs.readFileSync(indexPath, "utf8").split("\n")) {
|
|
if (phaseHeaderRe.test(line)) {
|
|
current = [];
|
|
phases.push(current);
|
|
continue;
|
|
}
|
|
const m = line.match(taskRowRe);
|
|
if (m && current) current.push(m[1]);
|
|
}
|
|
return phases.filter((phase) => phase.length > 0);
|
|
}
|
|
|
|
function phaseLabelFor(phaseTasks, index) {
|
|
const first = phaseTasks[0];
|
|
const id = typeof first === "string" ? first : first.id;
|
|
const dot = id.indexOf(".");
|
|
return dot === -1 ? `phase-${index}` : id.slice(0, dot);
|
|
}
|
|
|
|
function logProgress(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
|
|
// 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. If
|
|
// omitted, phases are discovered from the repo's own tasks/INDEX.md and run
|
|
// judgeOnly first (a cheap "is this already done" check against the
|
|
// board's possibly-stale checkmarks). Each phase gets its own branch
|
|
// (agent-run/<repoId>/<phaseLabel>, e.g. .../T1); once every task in that
|
|
// phase lands "done" or "done-with-concerns" AND the phase judge (the same
|
|
// pooled judge agent that reviewed each task) passes the integration
|
|
// review, the phase branch is squash-merged into baseBranch and pushed,
|
|
// then the next phase branches off that updated base. Any failure halts
|
|
// 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);
|
|
// repoId is a slug derived from the repo URL now (see slugFor), not a
|
|
// fresh UUID -- reusable across separate `runCoordinator` invocations
|
|
// against the same repo, so a stale clone from a prior run has to be
|
|
// wiped before this one starts, not merged into.
|
|
fs.rmSync(cwd, { recursive: true, force: true });
|
|
fs.mkdirSync(cwd, { recursive: true });
|
|
const pool = {};
|
|
|
|
const finish = (status) => {
|
|
pipelineSession.status = status;
|
|
pipelineSession.endedAt = new Date().toISOString();
|
|
logProgress(pipelineSession);
|
|
return pipelineSession;
|
|
};
|
|
|
|
// Deterministic, not routed through an LLM -- clone is 100% mechanical
|
|
// (same reasoning as commitPending/the squash-merge sequence below), and
|
|
// was the one place left that broke that pattern: a headless spawnPi
|
|
// call here meant a crash gave zero diagnostic output, just a silent
|
|
// exit code with nothing to debug from.
|
|
const clone = await runGit(cwd, ["clone", "--branch", baseBranch, repo, "."]);
|
|
if (clone.code !== 0) {
|
|
pipelineSession.gitError = clone.out;
|
|
return finish("clone-crashed");
|
|
}
|
|
if (!fs.existsSync(path.join(cwd, ".git"))) return finish("clone-missing");
|
|
|
|
let phases = tasks ? (Array.isArray(tasks[0]) ? tasks : [tasks]) : parseTaskBoard(cwd);
|
|
if (!phases || phases.length === 0) {
|
|
pipelineSession.gitError = "no tasks given and tasks/INDEX.md not found or empty";
|
|
return finish("no-tasks-found");
|
|
}
|
|
if (!tasks) {
|
|
phases = phases.map((phase) => phase.map((id) => ({ id, judgeOnly: true })));
|
|
}
|
|
pipelineSession.totalTasks = phases.flat().length;
|
|
|
|
for (let i = 0; i < phases.length; i++) {
|
|
const phaseTasks = phases[i];
|
|
const phaseLabel = phaseLabelFor(phaseTasks, i);
|
|
const phaseBranch = branchName ? `${branchName}/${phaseLabel}` : `agent-run/${repoId}/${phaseLabel}`;
|
|
|
|
const branchResult = await runGit(cwd, ["checkout", "-b", phaseBranch]);
|
|
if (branchResult.code !== 0) {
|
|
pipelineSession.gitError = branchResult.out;
|
|
return finish("branch-crashed");
|
|
}
|
|
|
|
if (i === 0) {
|
|
const gitignoreAdditions = [
|
|
"",
|
|
"# agent-harness: build artifacts and vendored archives never belong in source control",
|
|
"*.tar.gz",
|
|
"*.tgz",
|
|
"*.crate",
|
|
"*.zip",
|
|
"*.bin",
|
|
"*.whl",
|
|
"vendor/",
|
|
"node_modules/",
|
|
"",
|
|
"# agent-harness: task/phase completion sentinel files, harness bookkeeping only",
|
|
".task-result-*",
|
|
".phase-result-*",
|
|
".stage-done-*",
|
|
"",
|
|
"# agent-harness: PLAN.md is per-task planner scratch state, never a deliverable",
|
|
"PLAN.md",
|
|
].join("\n");
|
|
fs.appendFileSync(path.join(cwd, ".gitignore"), gitignoreAdditions + "\n");
|
|
await runGit(cwd, ["add", ".gitignore"]);
|
|
await runGit(cwd, ["commit", "-m", "chore: broaden .gitignore for agent-run artifacts"]);
|
|
}
|
|
|
|
await runPhase(cwd, phaseBranch, phaseTasks, pool, repoId, pipelineSession);
|
|
|
|
const phaseTaskIds = new Set(phaseTasks.map((t) => (typeof t === "string" ? t : t.id)));
|
|
const phaseResults = pipelineSession.taskResults.filter((r) => phaseTaskIds.has(r.task));
|
|
const phaseClean =
|
|
phaseResults.length === phaseTaskIds.size && phaseResults.every((r) => r.status === "done" || r.status === "done-with-concerns");
|
|
|
|
if (!phaseClean) {
|
|
pipelineSession.haltedAt = phaseLabel;
|
|
return finish("halted-phase-failed");
|
|
}
|
|
|
|
const phaseResultFile = path.join(cwd, `.phase-result-${phaseLabel}`);
|
|
fs.rmSync(phaseResultFile, { force: true });
|
|
const phaseJudge = await runOnPool(
|
|
pool,
|
|
cwd,
|
|
repoId,
|
|
"judge",
|
|
`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 ` +
|
|
`individual task already passed its own judge review -- your job here is different: confirm the ` +
|
|
`tasks integrate correctly as one coherent narrative, and that real integration tests (not just ` +
|
|
`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 ` +
|
|
`"VERDICT: FAIL" line plus rationale, then run: touch .stage-done-phase-${phaseLabel}-judge`,
|
|
path.join(cwd, `.stage-done-phase-${phaseLabel}-judge`)
|
|
);
|
|
await commitPending(cwd, `phase: ${phaseLabel} integration review`);
|
|
|
|
if (!phaseJudge.ok) {
|
|
pipelineSession.haltedAt = phaseLabel;
|
|
pipelineSession.gitError = phaseJudge.error;
|
|
return finish("phase-judge-crashed");
|
|
}
|
|
|
|
const phaseJudgeText = fs.existsSync(phaseResultFile) ? fs.readFileSync(phaseResultFile, "utf8") : "";
|
|
if (parseVerdictLine(phaseJudgeText, "VERDICT") !== "PASS") {
|
|
pipelineSession.haltedAt = phaseLabel;
|
|
pipelineSession.phaseJudgeRationale = phaseJudgeText;
|
|
return finish("halted-phase-judge-failed");
|
|
}
|
|
|
|
const checkoutBase = await runGit(cwd, ["checkout", baseBranch]);
|
|
if (checkoutBase.code !== 0) {
|
|
pipelineSession.gitError = checkoutBase.out;
|
|
return finish("squash-crashed");
|
|
}
|
|
const squash = await runGit(cwd, ["merge", "--squash", phaseBranch]);
|
|
if (squash.code !== 0) {
|
|
await runGit(cwd, ["merge", "--abort"]);
|
|
pipelineSession.gitError = squash.out;
|
|
return finish("squash-crashed");
|
|
}
|
|
const commit = await runGit(cwd, ["commit", "-m", `feat: ${phaseLabel} (${[...phaseTaskIds].join(", ")})`]);
|
|
if (commit.code !== 0) {
|
|
pipelineSession.gitError = commit.out;
|
|
return finish("squash-crashed");
|
|
}
|
|
const push = await runGit(cwd, ["push", "origin", baseBranch]);
|
|
if (push.code !== 0) {
|
|
pipelineSession.gitError = push.out;
|
|
return finish("squash-push-crashed");
|
|
}
|
|
logProgress(pipelineSession);
|
|
}
|
|
|
|
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). The backend
|
|
// (homelab-ornith) actually runs 2 GPU replicas behind one Kubernetes
|
|
// Service, each with its own copy of the model loaded (see homelab's
|
|
// k8s/apps/llm-serving/ornith.yaml) -- so up to 2 concurrent LLM calls get
|
|
// real independent instances; a 3rd+ concurrent call queues inside
|
|
// whichever replica the Service's own load-balancing lands it on (each
|
|
// replica runs OLLAMA_NUM_PARALLEL=1). REPO_CONCURRENCY above 2 is still
|
|
// useful (more repos in flight overlaps git/file work, not just LLM calls)
|
|
// but past 2 simultaneous LLM calls, extra concurrency mostly means queueing
|
|
// rather than added throughput -- bump the backend's replica count to
|
|
// change that, not this constant.
|
|
const REPO_CONCURRENCY = Number(process.env.REPO_CONCURRENCY) || 3;
|
|
|
|
// repoId is the repo's own name, not a random id -- it's what every role
|
|
// session's --name is built from (see runOnPool: `${repoId}-${role}`), so
|
|
// agent-manager's own session list groups naturally by repo ("portfolio-
|
|
// planner", "portfolio-judge", "poiman-planner", ...) instead of by opaque
|
|
// UUID. Takes the last path segment of the URL, strips a trailing `.git`,
|
|
// and sanitizes anything that isn't safe in a tmux session name / directory
|
|
// name / git branch name. Two different repos that happen to share a
|
|
// basename (e.g. two orgs' "portfolio") would collide -- not handled, since
|
|
// nothing about this harness's usage has needed more than one org per run.
|
|
function slugFor(repoUrl) {
|
|
const last = repoUrl.replace(/\/+$/, "").split("/").pop() || repoUrl;
|
|
return last.replace(/\.git$/, "").replace(/[^a-zA-Z0-9._-]/g, "-");
|
|
}
|
|
|
|
// 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 = slugFor(repoUrl);
|
|
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) {
|
|
const opts = { base: "main" };
|
|
for (let i = 0; i < argv.length; i++) {
|
|
const a = 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 === "--tasks") opts.tasks = argv[++i];
|
|
else if (a === "--branch") opts.branch = argv[++i];
|
|
}
|
|
return opts;
|
|
}
|
|
|
|
async function main() {
|
|
const opts = parseArgs(process.argv.slice(2));
|
|
const repos = opts.repos ? opts.repos.split(",") : opts.repo ? [opts.repo] : null;
|
|
if (!repos || repos.length === 0) {
|
|
console.error(
|
|
"usage: coordinator.js --repos <url1,url2,...> [--tasks T0.1,T0.2;T1.1,T1.2,...] [--base main] [--branch <name>]\n" +
|
|
" --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
|
|
// through kubectl exec (not a TTY) can drop buffered console.log/
|
|
// console.error output if the process exits before it flushes. Setting
|
|
// exitCode and letting the event loop drain naturally is the
|
|
// documented-safe way to exit with a specific code without racing it.
|
|
process.exitCode = 1;
|
|
return;
|
|
}
|
|
const phases = opts.tasks ? opts.tasks.split(";").map((phase) => phase.split(",")) : null;
|
|
const sessions = await runCoordinator({ repos, base: opts.base, tasks: phases, branchName: opts.branch });
|
|
process.exitCode = Object.values(sessions).every((s) => s.status === "completed") ? 0 : 1;
|
|
}
|
|
|
|
if (require.main === module) {
|
|
main();
|
|
}
|
|
|
|
module.exports = { runCoordinator, runRepoPipeline, runOnPool, parseTaskBoard };
|