Story Crater Bot
19cc4062b4
A pooled session's shell cwd drifts as it explores the repo between turns. Seen live: a repo whose internal workspace dir is one letter off from the repo's own directory name was enough for the agent to touch its sentinel one level off from where coordinator watches for it -- coordinator waited out the full timeout for a file that existed, just in the wrong place.
795 lines
39 KiB
YAML
795 lines
39 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];
|
|
|
|
// A pooled session's shell cwd drifts as it explores the repo (e.g. cd
|
|
// into a Rust workspace subdirectory to read source) and nothing resets
|
|
// it back between turns. Seen in practice: a repo whose own internal
|
|
// workspace folder is one letter off from the repo's own directory name
|
|
// ("poiman" the repo vs. "poimen" the crate workspace inside it) was
|
|
// enough for the agent to touch its sentinel one level off from where
|
|
// this function is watching for it -- coordinator waits out the full
|
|
// STAGE_TIMEOUT_MS for a file that already exists, just in the wrong
|
|
// place. State the absolute target directory and use absolute paths for
|
|
// every filesystem instruction, so there's nothing for the agent to get
|
|
// wrong by reasoning about a relative "current directory."
|
|
const cwdReminder = `Your working directory for this task is ${cwd} -- if your shell isn't already there, run: cd ${cwd}\n\n`;
|
|
|
|
if (!target) {
|
|
const spawnArgs = ["spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--prompt", cwdReminder + 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, cwdReminder + 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 run: touch ${sentinelFile}`, "Enter"]);
|
|
ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS);
|
|
}
|
|
}
|
|
return { ok, sessionName: label };
|
|
}
|
|
|
|
function plannerPrompt(task, specHint, judgeOnly, cwd) {
|
|
// 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 resultFile = path.join(cwd, `.task-result-${task}`);
|
|
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 ` +
|
|
`${resultFile} 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 ${path.join(cwd, `.stage-done-${task}-planner`)}`
|
|
);
|
|
}
|
|
|
|
function investigatorPrompt(task, cwd) {
|
|
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 ${path.join(cwd, `.stage-done-${task}-investigator`)}`
|
|
);
|
|
}
|
|
|
|
function implementerPrompt(task, attempt, feedbackHint, cwd) {
|
|
return (
|
|
`Use the implementer skill to implement what the current PLAN.md specifies for task ${task} (commit as you go). ` +
|
|
`${feedbackHint} Then run: touch ${path.join(cwd, `.stage-done-${task}-implementer-${attempt}`)}`
|
|
);
|
|
}
|
|
|
|
function judgePrompt(task, baseBranch, attempt, cwd) {
|
|
return (
|
|
`Use the judge skill to review the diff against ${baseBranch}...HEAD for task ${task}. Write your verdict to ` +
|
|
`${path.join(cwd, `.task-result-${task}`)} as a single "VERDICT: PASS" or "VERDICT: FAIL" line plus one line of rationale, then ` +
|
|
`run: touch ${path.join(cwd, `.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, cwd), 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, cwd), 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
|
|
? `${resultFile} 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 ${resultFile} -- read it and address every issue it raises.`
|
|
: "";
|
|
justRevisedPlan = false;
|
|
|
|
result = await stage(
|
|
"implementer",
|
|
implementerPrompt(task, implementAttempt, feedbackHint, cwd),
|
|
path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`)
|
|
);
|
|
if (!result.ok) return abandon("implementer", result, implementAttempt);
|
|
|
|
result = await stage("judge", judgePrompt(task, baseBranch, implementAttempt, cwd), 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 ${resultFile}, 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 ${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);
|
|
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 ${phaseResultFile} as a single "VERDICT: PASS" or ` +
|
|
`"VERDICT: FAIL" line plus rationale, then run: touch ${path.join(cwd, `.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 };
|