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. The difference is who owns each stage's
    // session: hub.js spawned raw tmux sessions itself and polled sentinel
    // files on a 10s timer to notice completion. This spawns every interactive
    // stage 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. The coordinator never
    // kills a stage's session; it rests at an idle prompt once its sentinel
    // file lands, and agent-manager's session list becomes the audit trail of
    // everything the pipeline ran. Completion is still signaled by sentinel
    // files under the task's worktree (unchanged convention), but 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 (runStage) --
    // 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 (both per-task and phase-judge -- both spawn with stageLabel
    // "judge", see runTaskInteractive/runPipeline) 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, same as before this existed.
    const JUDGE_PROVIDER = process.env.JUDGE_PROVIDER || AGENT_PROVIDER;
    const JUDGE_MODEL = process.env.JUDGE_MODEL || AGENT_MODEL;
    
    function providerModelFor(stageLabel) {
      return stageLabel === "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 the RETRY nudge, 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"]);
    
    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 (runStage,
    // below) move to 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, task, diagnosticPrompt) {
      const result = await spawnPi({ agent: "resolver", prompt: diagnosticPrompt, cwd });
      return parseVerdictLine(result.lastText, "RESOLUTION");
    }
    
    function taskSessionLabel(task) {
      return `task-${task.replace(/[^a-zA-Z0-9]/g, "-")}`;
    }
    
    // POSIX single-quote wrap, for embedding a literal `agent-manager spawn`
    // command inside another stage's prompt -- the same escaping tmux.ShellQuote
    // does on the Go side of the fork.
    function shQuote(s) {
      return "'" + String(s).replace(/'/g, `'\\''`) + "'";
    }
    
    // The literal shell command an agent runs, as its own last action, to spawn
    // its successor stage itself instead of the coordinator spawning it. Always
    // applies HARD_RULES to the embedded prompt -- the single place that
    // happens for agent-issued spawns, mirroring spawnRole's HARD_RULES + prompt
    // for coordinator-issued ones (see runTaskInteractive) so it's applied
    // exactly once regardless of who calls agent-manager spawn.
    function spawnCommandLine(cwd, task, stageLabel, prompt) {
      const label = `${taskSessionLabel(task)}-${stageLabel}`;
      const parts = [AGENT_MANAGER_BIN, "spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--prompt", HARD_RULES + prompt];
      const { provider, model } = providerModelFor(stageLabel);
      if (provider) parts.push("--provider", provider);
      if (model) parts.push("--model", model);
      return parts.map(shQuote).join(" ");
    }
    
    // Instruction builders for the self-chained happy path: planner spawns
    // investigator spawns implementer spawns judge, each as its own last
    // action. Built bottom-up (judge first) since each level embeds the next
    // level's full spawn command as literal text. Judge never self-spawns --
    // the coordinator reads its verdict and decides PASS/retry/revise itself
    // (see runTaskInteractive), so retries and plan revisions stay a
    // deterministic, hard-capped branch in code, not something an agent
    // decides on its own.
    function judgeInstructions(task, baseBranch, attempt) {
      return (
        `Use the judge skill to review the diff against ${baseBranch}...HEAD. Write your verdict to ` +
        `.task-result-${task} as a single "VERDICT: PASS" or "VERDICT: FAIL" line plus one line of ` +
        `rationale, then run: touch .stage-done-${task}-judge-${attempt}`
      );
    }
    
    function implementerInstructions(cwd, task, baseBranch, attempt, feedbackHint) {
      const judgeSpawn = spawnCommandLine(cwd, task, "judge", judgeInstructions(task, baseBranch, attempt));
      return (
        `Use the implementer skill to implement what the current PLAN.md specifies (commit as you go). ${feedbackHint} ` +
        `When your implementation is committed, spawn the judge stage yourself by running exactly this command: ${judgeSpawn}\n\n` +
        `Then run: touch .stage-done-${task}-implementer-${attempt}`
      );
    }
    
    function investigatorInstructions(cwd, task, baseBranch) {
      const implSpawn = spawnCommandLine(cwd, task, "implementer", implementerInstructions(cwd, task, baseBranch, 1, ""));
      return (
        `Use the investigator skill to confirm PLAN.md against real sources, append findings, commit. ` +
        `When finished, spawn the implementer stage yourself by running exactly this command: ${implSpawn}\n\n` +
        `Then run: touch .stage-done-${task}-investigator`
      );
    }
    
    function plannerInstructions(cwd, task, baseBranch, specHint) {
      const invSpawn = spawnCommandLine(cwd, task, "investigator", investigatorInstructions(cwd, task, baseBranch));
      return (
        `Use the planner skill to draft PLAN.md for task ${task}, reading its spec (${specHint}). Commit PLAN.md. ` +
        `When finished, spawn the investigator stage yourself by running exactly this command: ${invSpawn}\n\n` +
        `Then run: touch .stage-done-${task}-planner`
      );
    }
    
    async function runConcurrent(items, limit, worker) {
      const results = new Array(items.length);
      let i = 0;
      async function next() {
        while (i < items.length) {
          const idx = i++;
          results[idx] = await worker(items[idx], idx);
        }
      }
      await Promise.all(Array.from({ length: Math.min(limit, items.length) }, next));
      return results;
    }
    
    const STAGE_TIMEOUT_MS = 10 * 60 * 1000;
    const 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 -- the "notify" half of the migration:
    // 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 role as its own fresh interactive pi session, spawned as a real
    // agent-manager-tracked session (see AGENT_MANAGER_BIN above) instead of a
    // raw tmux pane. Planner, investigator, implementer, judge, and
    // phase-judge are separate agents with separate context, coordinating only
    // through what's on disk in the task's worktree -- unchanged. The session
    // is never killed here: it rests at an idle prompt once its sentinel file
    // lands, staying attachable in agent-manager for as long as the user wants.
    //
    // Used for stages the coordinator itself spawns: the very first stage of a
    // task (planner), implementer retries and planner-revise (both are the
    // coordinator's own branch decision after reading judge's verdict -- see
    // runTaskInteractive), and phase-judge. Stages self-spawned by their
    // predecessor (see spawnCommandLine and the instruction builders above) go
    // through waitForStage below instead, since the coordinator never issued
    // the spawn itself.
    async function spawnAndWaitStage(cwd, task, stageLabel, stagePrompt, sentinelFile) {
      const label = `${taskSessionLabel(task)}-${stageLabel}`;
      fs.rmSync(sentinelFile, { force: true });
    
      const spawnArgs = ["spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--prompt", stagePrompt];
      const { provider, model } = providerModelFor(stageLabel);
      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 };
      const sessionId = spawned.out;
      const target = amSessionName(sessionId);
    
      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,
          task,
          `Task ${task}'s "${stageLabel}" stage hasn't finished 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 the "${stageLabel}" stage now and touch ${path.basename(sentinelFile)} when done.`,
            "Enter",
          ]);
          ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS);
        }
      }
      return { ok, sessionName: label, sessionId };
    }
    
    // Waits for a stage that spawned itself (its predecessor's prompt embedded
    // the agent-manager spawn command -- see spawnCommandLine). No session id
    // is available here for pane capture on a timeout, since the coordinator
    // never issued the spawn; the resolver gets a diagnostic prompt pointing at
    // the session name to attach to directly instead, and RETRY just gives it
    // the nudge window rather than an actual send-keys nudge.
    async function waitForStage(cwd, task, stageLabel, sentinelFile) {
      const label = `${taskSessionLabel(task)}-${stageLabel}`;
      let ok = await waitForSentinel(sentinelFile, STAGE_TIMEOUT_MS);
      if (!ok) {
        const resolution = await askResolver(
          cwd,
          task,
          `Task ${task}'s "${stageLabel}" stage hasn't finished after 10 minutes. It was self-spawned by its ` +
            `predecessor agent -- attach to its agent-manager session ("${label}") directly to inspect progress; no ` +
            `pane tail is available here. Decide: is it still making real progress and worth waiting longer, or stuck ` +
            `and worth abandoning?`
        );
        if (resolution === "RETRY") {
          ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS);
        }
      }
      return { ok, sessionName: label };
    }
    
    const MAX_IMPLEMENT_ATTEMPTS = 5;
    const MAX_PLAN_REVISIONS = 3;
    
    // Runs one task in its own git worktree (see runPhase): planner drafts
    // PLAN.md, 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, planner is brought back in to
    // judge whether the *plan* itself is wrong, not just the implementation; if
    // so it revises PLAN.md and the implementer gets a fresh attempt budget
    // against the new plan. MAX_PLAN_REVISIONS caps this from looping forever
    // on a task that's genuinely stuck.
    async function runTaskInteractive(cwd, baseBranch, task, pipelineSession, judgeOnly) {
      const resultFile = path.join(cwd, `.task-result-${task}`);
      fs.rmSync(resultFile, { force: true });
    
      const specHint = `the file under tasks/ starting with "${task}-"`;
    
      // spawnRole: coordinator issues the spawn itself (planner's first call,
      // implementer retries, planner-revise, the judgeOnly quick check).
      // waitRole: the stage was self-spawned by its predecessor (see
      // spawnCommandLine/instruction builders above) -- the coordinator only
      // waits. Both still run the commitPending backstop sweep afterward and
      // update pipelineSession's progress display, unchanged from before.
      const spawnRole = async (stageLabel, prompt, sentinel) => {
        pipelineSession.activeTasks[task] = { stage: stageLabel, startedAt: new Date().toISOString() };
        logProgress(pipelineSession);
        const result = await spawnAndWaitStage(cwd, task, stageLabel, HARD_RULES + prompt, sentinel);
        await commitPending(cwd, `task: ${task} (${stageLabel})`);
        return result;
      };
    
      const waitRole = async (stageLabel, sentinel) => {
        pipelineSession.activeTasks[task] = { stage: stageLabel, startedAt: new Date().toISOString() };
        logProgress(pipelineSession);
        const result = await waitForStage(cwd, task, stageLabel, sentinel);
        await commitPending(cwd, `task: ${task} (${stageLabel})`);
        return result;
      };
    
      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 } : {}),
        };
      };
    
      if (judgeOnly) {
        const quick = await spawnRole(
          "judge",
          `Task ${task} may already be implemented on this branch -- check ` +
            `\`git log --oneline --grep '${task}'\` and the current code directly against its spec ` +
            `(${specHint})'s acceptance criteria (no PLAN.md exists for this task yet). 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-0`,
          path.join(cwd, `.stage-done-${task}-judge-0`)
        );
        if (!quick.ok) return abandon("judge", quick, 0);
    
        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 };
        }
      }
    
      // Happy path is self-chained: planner spawns investigator spawns
      // implementer (attempt 1) spawns judge, each as its own last action (see
      // the instruction builders above). The coordinator spawns only planner
      // here, then just waits for each stage's sentinel in turn -- same
      // per-stage timeout/resolver/commit backstop as before, just not the one
      // issuing the spawn for the self-chained stages.
      let result = await spawnRole("planner", plannerInstructions(cwd, task, baseBranch, specHint), path.join(cwd, `.stage-done-${task}-planner`));
      if (!result.ok) return abandon("planner", result);
    
      result = await waitRole("investigator", 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;
      // Attempt 1's implementer was already self-spawned by investigator above
      // (its exact prompt was pre-baked into investigatorInstructions with
      // attempt=1 and an empty feedbackHint, matching what attempt 1 always
      // computes below) -- the coordinator only waits for it. Every retry
      // after that is the coordinator's own branch decision on judge's
      // verdict, so it spawns implementer directly instead.
      let implementerSelfChained = true;
    
      while (true) {
        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;
    
        if (implementerSelfChained) {
          result = await waitRole("implementer", path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`));
        } else {
          result = await spawnRole(
            "implementer",
            implementerInstructions(cwd, task, baseBranch, implementAttempt, feedbackHint),
            path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`)
          );
        }
        implementerSelfChained = false;
        if (!result.ok) return abandon("implementer", result, implementAttempt);
    
        // Judge is always self-spawned by implementer, coordinator- or
        // self-chained alike -- implementerInstructions embeds the same spawn
        // command either way.
        result = await waitRole("judge", path.join(cwd, `.stage-done-${task}-judge-${implementAttempt}`));
        if (!result.ok) return abandon("judge", result, implementAttempt);
    
        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 spawnRole(
            "planner-revise",
            `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 and ` +
              `commit. If you change the approach, also use the investigator skill to confirm the new approach against ` +
              `real sources before committing. If the plan is sound, note why in PLAN.md and leave it as-is. Then run: ` +
              `touch .stage-done-${task}-planner-revise-${planRevisions}`,
            path.join(cwd, `.stage-done-${task}-planner-revise-${planRevisions}`)
          );
          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 };
    }
    
    // How many tasks' independent chains can be mid-flight at once within a
    // phase. This is the real buffer against a single-instance backend: a
    // lock around agent-manager spawn wouldn't work (spawn is fire-and-forget
    // -- it creates the tmux session and returns immediately, the actual LLM
    // call happens later, asynchronously, inside that detached session, so a
    // lock held only for the spawn call releases before the call it's
    // supposed to guard even starts). Ollama itself queues concurrent
    // requests to one loaded model rather than erroring, but each one waits
    // out the others -- PHASE_CONCURRENCY=1 makes that explicit: tasks run
    // strictly one after another, one chain ever active against the backend
    // at a time. Bump it back up once there's more than one model instance.
    const PHASE_CONCURRENCY = Number(process.env.PHASE_CONCURRENCY) || 3;
    
    // Runs one phase (a batch of tasks with no declared dependency on each
    // other) with up to PHASE_CONCURRENCY tasks in flight at once. Each task
    // gets its own git worktree off workBranch -- concurrent pi sessions writing
    // into one shared working tree would corrupt the index; worktrees share the
    // same object database but give each task an isolated checkout. After a
    // task's session ends, its branch is merged back into workBranch and pushed,
    // one merge at a time (git ref updates aren't safe to run concurrently even
    // though the worktrees themselves are isolated).
    async function runPhase(pipelineId, cwd, workBranch, phaseTasks, pipelineSession) {
      const entries = phaseTasks.map((t) => (typeof t === "string" ? { id: t, judgeOnly: false } : t));
    
      const worktrees = {};
      for (const entry of entries) {
        const task = entry.id;
        const wtDir = path.join(WORK_DIR, pipelineId, `wt-${task.replace(/[^a-zA-Z0-9]/g, "-")}`);
        const taskBranch = `task/${task}`;
        const add = await runGit(cwd, ["worktree", "add", "-b", taskBranch, wtDir, workBranch]);
        if (add.code !== 0) {
          pipelineSession.taskResults.push({ task, status: "worktree-crashed", error: add.out });
          continue;
        }
        worktrees[task] = { wtDir, taskBranch };
      }
    
      const runnable = entries.filter((e) => worktrees[e.id]);
      await runConcurrent(runnable, PHASE_CONCURRENCY, async (entry) => {
        const { wtDir } = worktrees[entry.id];
        const result = await runTaskInteractive(wtDir, workBranch, entry.id, pipelineSession, entry.judgeOnly);
        pipelineSession.taskResults.push(result);
        return result;
      });
    
      for (const entry of runnable) {
        const task = entry.id;
        const { wtDir, taskBranch } = worktrees[task];
        const result = pipelineSession.taskResults.find((r) => r.task === task);
    
        const merge = await runGit(cwd, ["merge", "--no-ff", taskBranch, "-m", `merge: ${task}`]);
        if (merge.code !== 0) {
          await runGit(cwd, ["merge", "--abort"]);
          if (result) {
            result.status = "merge-conflict";
            result.mergeError = merge.out;
          }
        } else {
          const push = await runGit(cwd, ["push", "-u", "origin", workBranch]);
          if (result) {
            result.pushed = push.code === 0;
            if (!result.pushed) result.pushError = push.out;
          }
        }
    
        await runGit(cwd, ["worktree", "remove", wtDir, "--force"]);
        await runGit(cwd, ["branch", "-D", taskBranch]);
        logProgress(pipelineSession);
      }
    }
    
    // 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(`[pipeline ${pipelineSession.id}] ${JSON.stringify(pipelineSession)}`);
    }
    
    // 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.
    // Phases run strictly sequentially; tasks within a phase run concurrently,
    // each in its own worktree -- see runPhase. Each phase gets its own branch
    // (agent-run/<id>/<phaseLabel>, e.g. .../T1); once every task in that
    // phase lands "done" or "done-with-concerns" AND the phase judge 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 the pipeline before merging.
    async function runPipeline({ pipelineId, repo, baseBranch, tasks, branchName }) {
      const cwd = path.join(WORK_DIR, pipelineId);
      fs.mkdirSync(cwd, { recursive: true });
    
      const pipelineSession = {
        id: pipelineId,
        status: "running",
        taskResults: [],
        activeTasks: {},
        totalTasks: 0,
        startedAt: new Date().toISOString(),
      };
    
      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");
    
      // No --tasks given: discover the phase/task board from the repo's own
      // tasks/INDEX.md (see parseTaskBoard) instead of requiring the caller to
      // already know it. Each discovered task runs judgeOnly first -- a cheap
      // "is this already done on this branch" check against the board's own
      // possibly-stale checkmarks, falling through to the full planner/
      // investigator/implementer/judge flow only when it isn't.
      let phases = tasks ? (Array.isArray(tasks[0]) ? tasks : [tasks]) : parseTaskBoard(cwd);
      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/${pipelineId}/${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-*",
          ].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(pipelineId, cwd, phaseBranch, phaseTasks, 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 phaseSentinel = path.join(cwd, `.stage-done-phase-${phaseLabel}-judge`);
        const phaseJudge = await spawnAndWaitStage(
          cwd,
          `phase-${phaseLabel}`,
          "judge",
          HARD_RULES +
            `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`,
          phaseSentinel
        );
        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");
    }
    
    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 === "--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));
      if (!opts.repo) {
        console.error(
          "usage: coordinator.js --repo <url> [--tasks T0.1,T0.2;T1.1,T1.2,...] [--base main] [--branch <name>]\n" +
            "  --tasks omitted: discovers phases from the cloned repo's own tasks/INDEX.md"
        );
        // 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 pipelineId = require("node:crypto").randomUUID();
      const result = await runPipeline({ pipelineId, repo: opts.repo, baseBranch: opts.base, tasks: phases, branchName: opts.branch });
      process.exitCode = result.status === "completed" ? 0 : 1;
    }
    
    if (require.main === module) {
      main();
    }
    
    module.exports = { runPipeline, spawnCommandLine, spawnAndWaitStage, waitForStage, parseTaskBoard };