homelab/k8s/apps/agent-pod/coordinator-configmap.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, "--group", repoId, "--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 });
      }
    }
    
    // Committed (never gitignored) so it survives a resumed phase branch --
    // one task id per line, appended as each task resolves. This is what lets
    // a resumed run skip straight past already-resolved tasks instead of
    // re-running planner's judgeOnly decision on every one of them again:
    // resuming the git branch alone only recovers the CODE, not "which tasks
    // are already settled," and re-deciding that from scratch for every task
    // burns a full LLM call per already-done task before ever reaching the
    // first one that actually needs work.
    function progressLedgerPath(cwd) {
      return path.join(cwd, ".agent-progress");
    }
    
    function readCompletedTasks(cwd) {
      const file = progressLedgerPath(cwd);
      if (!fs.existsSync(file)) return new Set();
      return new Set(
        fs
          .readFileSync(file, "utf8")
          .split("\n")
          .map((line) => line.trim())
          .filter(Boolean)
      );
    }
    
    async function recordTaskComplete(cwd, task) {
      fs.appendFileSync(progressLedgerPath(cwd), `${task}\n`);
      await runGit(cwd, ["add", path.basename(progressLedgerPath(cwd))]);
      await runGit(cwd, ["commit", "-m", `chore: mark ${task} complete in progress ledger`]);
    }
    
    // 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. baseBranch here is the TRUE base (e.g. "main") -- judge
    // reviews `git diff baseBranch...HEAD`, not phaseBranch...HEAD, which would
    // always be empty since HEAD *is* phaseBranch while it's checked out.
    //
    // Pushes phaseBranch after every task, not just once at full-phase-end: the
    // pod is ephemeral and every restart re-clones baseBranch fresh (see
    // runRepoPipeline) -- without this, a redeploy mid-phase silently discards
    // every task committed so far, and the next run re-decides "is this done?"
    // from a clone that never saw any of that work.
    async function runPhase(cwd, baseBranch, phaseBranch, phaseTasks, pool, repoId, pipelineSession) {
      const entries = phaseTasks.map((t) => (typeof t === "string" ? { id: t, judgeOnly: false } : t));
      const completed = readCompletedTasks(cwd);
    
      for (const entry of entries) {
        if (completed.has(entry.id)) {
          const result = { task: entry.id, status: "done", resumed: true };
          pipelineSession.taskResults.push(result);
          logProgress(pipelineSession);
          continue;
        }
    
        const result = await runTaskOnPool(cwd, baseBranch, entry.id, pool, repoId, pipelineSession, entry.judgeOnly);
        pipelineSession.taskResults.push(result);
        if (result.status === "done" || result.status === "done-with-concerns") {
          await recordTaskComplete(cwd, entry.id);
        }
        await runGit(cwd, ["push", "-u", "origin", phaseBranch]);
        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}`;
    
        // Resume a phase branch a prior (since-restarted) run already pushed,
        // instead of always branching fresh off baseBranch -- otherwise every
        // redeploy silently discards whatever tasks that prior run already
        // committed and pushed (see runPhase's per-task push below).
        const fetchExisting = await runGit(cwd, ["fetch", "origin", phaseBranch]);
        const resuming = fetchExisting.code === 0;
        const branchResult = resuming
          ? await runGit(cwd, ["checkout", "-b", phaseBranch, "FETCH_HEAD"])
          : await runGit(cwd, ["checkout", "-b", phaseBranch]);
        if (branchResult.code !== 0) {
          pipelineSession.gitError = branchResult.out;
          return finish("branch-crashed");
        }
    
        if (i === 0 && !resuming) {
          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, baseBranch, 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");
        }
    
        // Milestone's content now lives in baseBranch as one squashed commit --
        // the phase branch (and whatever a prior restart already pushed of it)
        // has no further reason to exist. Delete it both places so a future run
        // never tries to resume a phase that's already done, and so origin
        // doesn't accumulate one dangling branch per completed phase forever.
        await runGit(cwd, ["branch", "-D", phaseBranch]);
        await runGit(cwd, ["push", "origin", "--delete", phaseBranch]);
    
        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 };