homelab/k8s/apps/agent-pod/hub-configmap.yaml

apiVersion: v1
data:
  hub.js: |
    #!/usr/bin/env node
    // agent-hub: lives inside the pi container (not a sidecar) so it can spawn
    // `pi` directly, and control the pod's own tmux server. One persistent
    // in-cluster service -- POST /run to trigger a single ad-hoc headless agent
    // run, POST /pipeline to run an ordered list of task phases against a
    // repo/branch (phases run sequentially, up to PHASE_CONCURRENCY tasks within
    // a phase run concurrently, each in its own git worktree). Within one task,
    // planner/investigator/implementer/judge are separate agents in separate
    // named tmux sessions (task-<id>-<role>, attachable via `kubectl exec -it --
    // tmux attach -t <name>` while running), coordinating only through what's on
    // disk in that task's worktree -- not one shared conversation. GET /console
    // (WebSocket) watches every concurrent headless run live, relaying pi's own
    // session protocol verbatim (same event shape Claude Code sessions use).
    const http = require("node:http");
    const crypto = require("node:crypto");
    const fs = require("node:fs");
    const path = require("node:path");
    const { spawn } = require("node:child_process");
    const readline = require("node:readline");
    const { WebSocketServer } = require("ws");

    const PORT = process.env.HUB_PORT || 9090;
    const WORK_DIR = process.env.HUB_WORK_DIR || path.join(require("node:os").tmpdir(), "agent-harness-work");

    // Never rely on a bare `pi` on $PATH -- both `pi` and `agent-console` collide
    // with unrelated tools on this machine (a Rust CLI and a Datadog TUI,
    // respectively, discovered the hard way this session). Always invoke the
    // exact pinned @earendil-works/[email protected] installed locally under
    // .pi-cli/, by explicit path.
    const PI_BIN =
      process.env.PI_BIN ||
      path.join(
        __dirname,
        "..",
        ".pi-cli",
        "node_modules",
        "@earendil-works",
        "pi-coding-agent",
        "dist",
        "cli.js"
      );

    // Job-type skills under pi/skills/<name>/SKILL.md (mounted at
    // ~/.pi/agent/skills/<name>/ in agent-pod). When `agent` matches one of
    // these, the prompt is forced through pi's `/skill:<name> <args>` mechanism
    // instead of being sent bare -- see pi's skills.md docs on single-shot skill
    // loading. `resolver` is never dispatched directly by a caller; only the
    // pipeline driver invokes it, on stage crashes.
    const ROLE_SKILLS = new Set([
      "planner",
      "investigator",
      "info-collector",
      "implementer",
      "judge",
      "resolver",
    ]);

    const sessions = new Map(); // id -> {id, agent, status, events, startedAt, endedAt, pipelineId?, stage?}
    const viewers = new Set(); // WebSocket connections watching /console

    function broadcast(type, session) {
      const msg = JSON.stringify({ type, session });
      for (const ws of viewers) {
        if (ws.readyState === ws.OPEN) ws.send(msg);
      }
    }

    function startSession(agent, extra = {}) {
      const id = extra.id || crypto.randomUUID();
      const session = {
        ...extra,
        id,
        agent,
        status: "running",
        events: [],
        startedAt: new Date().toISOString(),
      };
      sessions.set(id, session);
      broadcast("start", session);
      return session;
    }

    function addEvent(session, rawLine) {
      const event = JSON.parse(rawLine);
      session.events.push(event);
      broadcast("event", session);
      return event;
    }

    function endSession(session, status) {
      session.status = status;
      session.endedAt = new Date().toISOString();
      broadcast("end", session);
    }

    // Extracts the plain-text content of a message_end event, if any -- used to
    // find the VERDICT:/RESOLUTION: line judge/resolver skills are required to
    // end their final message with.
    function textOf(event) {
      if (event.type !== "message_end" || !event.message || !Array.isArray(event.message.content)) {
        return null;
      }
      return event.message.content
        .filter((c) => c.type === "text")
        .map((c) => c.text)
        .join("\n");
    }

    // Core spawn primitive. Spawns `pi -p --mode json <extraArgs> <prompt>`,
    // relays every line as a session event exactly like before. Returns
    // { session, done } -- `session` is available synchronously (so an HTTP
    // handler can respond with its id right away, same as the old runAgent),
    // `done` is a Promise resolving once the process exits, for callers that
    // need to wait on a stage (the pipeline driver) rather than fire-and-forget.
    function spawnPi({ agent, prompt, provider, model, cwd, sessionExtra = {} }) {
      const session = startSession(agent, sessionExtra);
      const args = ["-p", "--mode", "json"];
      if (provider) args.push("--provider", provider);
      if (model) args.push("--model", model);
      const finalPrompt = ROLE_SKILLS.has(agent) ? `/skill:${agent} ${prompt}` : prompt;
      args.push(finalPrompt);

      const child = spawn(PI_BIN, args, {
        stdio: ["ignore", "pipe", "pipe"],
        cwd,
      });
      const rl = readline.createInterface({ input: child.stdout });
      let lastText = "";
      let stderrTail = "";

      rl.on("line", (line) => {
        if (!line.trim()) return;
        try {
          const event = addEvent(session, line);
          const text = textOf(event);
          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);
      });

      const done = new Promise((resolve) => {
        child.on("close", (code) => {
          endSession(session, code === 0 ? "done" : "error");
          resolve({ code, session, lastText, stderrTail });
        });
      });

      return { session, done };
    }

    function runAgent(agent, prompt, extraArgs = {}) {
      // Fire-and-forget: caller (the /run handler) doesn't await `done`.
      return spawnPi({ agent, prompt, ...extraArgs }).session;
    }

    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;
    }

    // Deterministic git operations, run directly by hub.js rather than left to
    // the model -- branch creation and pushing after each task are mechanical,
    // not judgment calls, and need to happen reliably every time regardless of
    // what a task's stages did or didn't remember to do.
    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 };
    }

    // Invokes the `resolver` skill to diagnose a stuck/crashed stage and decide
    // RETRY vs ABORT. Shared by both crash-recovery paths below (headless
    // exit-code failures and interactive sentinel-file timeouts) -- the
    // diagnostic prompt differs per caller, but "ask resolver, parse the
    // RESOLUTION: line" is identical either way.
    async function askResolver(pipelineId, cwd, task, diagnosticPrompt) {
      const resolverResult = await spawnPi({
        agent: "resolver",
        prompt: diagnosticPrompt,
        cwd,
        sessionExtra: { pipelineId, stage: "resolver", task },
      }).done;
      return parseVerdictLine(resolverResult.lastText, "RESOLUTION");
    }

    // Runs one pipeline stage, and if it crashes (nonzero exit -- not a semantic
    // judge FAIL, which is handled separately), asks the resolver to diagnose
    // and decide RETRY vs ABORT. Retries the failed stage at most once,
    // regardless of what resolver recommends a second time -- a hard cap, not
    // indefinite trust in the model's judgment.
    async function runStageWithResolver(pipelineId, cwd, stage, prompt, task) {
      let result = await spawnPi({
        agent: stage,
        prompt,
        cwd,
        sessionExtra: { pipelineId, stage, task },
      }).done;
      if (result.code === 0) return result;

      const resolution = await askResolver(
        pipelineId,
        cwd,
        task,
        `Stage "${stage}" exited with code ${result.code}. Its stderr tail:\n${result.stderrTail}`
      );

      if (resolution === "RETRY") {
        result = await spawnPi({
          agent: stage,
          prompt,
          cwd,
          sessionExtra: { pipelineId, stage, task },
        }).done;
      }
      return result;
    }

    // Deterministic tmux operations -- same rationale as runGit: mechanical,
    // not a judgment call, run directly rather than trusted to a prompt.
    function runTmux(args) {
      return runCmd("tmux", args);
    }

    function tmuxSessionName(task) {
      return `task-${task.replace(/[^a-zA-Z0-9]/g, "-")}`;
    }

    // Bounded-concurrency pool -- runs `worker` over `items`, at most `limit` in
    // flight at once. No external dep; a plain in-order index cursor shared by
    // `limit` runner loops.
    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;
    const POLL_MS = 10 * 1000;

    async function waitForFile(filePath, limitMs) {
      const start = Date.now();
      while (!fs.existsSync(filePath)) {
        if (Date.now() - start > limitMs) return false;
        await new Promise((r) => setTimeout(r, POLL_MS));
      }
      return true;
    }

    const MAX_IMPLEMENT_ATTEMPTS = 5;
    const MAX_PLAN_REVISIONS = 3;

    // Runs one role as its own fresh interactive pi session in its own named
    // tmux session -- planner, investigator, implementer, and judge are
    // separate agents with separate context, not turns in one shared
    // conversation. They coordinate only through what's on disk in the task's
    // worktree: PLAN.md, committed code, judge's result file. Each session is
    // attachable while it runs (kubectl exec -it -- tmux attach -t <name>) and
    // killed once its sentinel file lands or it's abandoned after resolver
    // escalation.
    async function runStage(pipelineId, cwd, task, stageLabel, stagePrompt, sentinelFile) {
      const sessionName = `${tmuxSessionName(task)}-${stageLabel}`;
      fs.rmSync(sentinelFile, { force: true });

      const spawned = await runTmux(["new-session", "-d", "-s", sessionName, "-c", cwd, PI_BIN, stagePrompt]);
      if (spawned.code !== 0) return { ok: false, crashed: true, error: spawned.out, sessionName };

      let ok = await waitForFile(sentinelFile, STAGE_TIMEOUT_MS);
      if (!ok) {
        const pane = await runTmux(["capture-pane", "-t", sessionName, "-p", "-S", "-200"]);
        const resolution = await askResolver(
          pipelineId,
          cwd,
          task,
          `Task ${task}'s "${stageLabel}" stage hasn't finished after 10 minutes. ` +
            `Its pane tail:\n${pane.out.slice(-3000)}\n\nDecide: is it still making ` +
            `real progress and worth nudging to wrap up, or stuck and worth abandoning?`
        );
        if (resolution === "RETRY") {
          await runTmux([
            "send-keys",
            "-t",
            sessionName,
            `Please wrap up the "${stageLabel}" stage now and touch ${path.basename(sentinelFile)} when done.`,
            "Enter",
          ]);
          ok = await waitForFile(sentinelFile, NUDGE_TIMEOUT_MS);
        }
      }

      await runTmux(["kill-session", "-t", sessionName]);
      return { ok, sessionName };
    }

    // 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(pipelineId, 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}-"`;

      const runRole = async (stageLabel, prompt, sentinel) => {
        pipelineSession.activeTasks[task] = {
          stage: stageLabel,
          sessionName: `${tmuxSessionName(task)}-${stageLabel}`,
          startedAt: new Date().toISOString(),
        };
        broadcast("event", pipelineSession);
        const result = await runStage(pipelineId, cwd, task, stageLabel, prompt, sentinel);
        await commitPending(cwd, `task: ${task} (${stageLabel})`);
        return result;
      };

      const abandon = (stageLabel, result, attempt) => {
        delete pipelineSession.activeTasks[task];
        broadcast("event", pipelineSession);
        return {
          task,
          status: result.crashed ? "spawn-crashed" : "timed-out",
          error: result.error,
          stoppedAt: stageLabel,
          ...(attempt !== undefined ? { attempt } : {}),
        };
      };

      // Task's implementation is inherited already-committed (e.g. from a base
      // branch of prior work) -- try one judge pass against the spec directly
      // (no PLAN.md exists yet) before paying for a full planner/investigator
      // redo. PASS ends the task here; FAIL falls through into the normal flow
      // below, so planner/implementer pick up with the judge's real feedback.
      if (judgeOnly) {
        const quick = await runRole(
          "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];
          broadcast("event", pipelineSession);
          return { task, status: "done", judgeRationale: quickText, judgeOnlyPass: true };
        }
      }

      let result = await runRole(
        "planner",
        `Use the planner skill to draft PLAN.md for task ${task}, reading its spec (${specHint}). Commit PLAN.md, then run: touch .stage-done-${task}-planner`,
        path.join(cwd, `.stage-done-${task}-planner`)
      );
      if (!result.ok) return abandon("planner", result);

      result = await runRole(
        "investigator",
        `Use the investigator skill to confirm PLAN.md against real sources, append findings, commit. Then run: touch .stage-done-${task}-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;

      while (true) {
        implementAttempt++;
        const feedbackHint = fs.existsSync(resultFile)
          ? justRevisedPlan
            ? `.task-result-${task} holds the judge's feedback against the OLD plan, which prompted a plan revision -- ` +
              `PLAN.md has since changed. Read the current PLAN.md as the source of truth, not the old feedback verbatim.`
            : `A previous judge review exists at .task-result-${task} -- read it and address every issue it raises.`
          : "";
        justRevisedPlan = false;

        result = await runRole(
          "implementer",
          `Use the implementer skill to implement what the current PLAN.md specifies (commit as you go). ${feedbackHint} Then run: touch .stage-done-${task}-implementer-${implementAttempt}`,
          path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`)
        );
        if (!result.ok) return abandon("implementer", result, implementAttempt);

        result = await runRole(
          "judge",
          `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-${implementAttempt}`,
          path.join(cwd, `.stage-done-${task}-judge-${implementAttempt}`)
        );
        if (!result.ok) return abandon("judge", result, implementAttempt);

        resultText = fs.existsSync(resultFile) ? fs.readFileSync(resultFile, "utf8") : "";
        verdict = parseVerdictLine(resultText, "VERDICT");
        if (verdict === "PASS") break;

        if (implementAttempt >= MAX_IMPLEMENT_ATTEMPTS) {
          if (planRevisions >= MAX_PLAN_REVISIONS) break;
          planRevisions++;
          result = await runRole(
            "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];
      broadcast("event", 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,
      };
    }

    const 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) {
      // Each entry is either a plain task id, or { id, judgeOnly: true } when
      // the task's implementation already exists (e.g. inherited from a base
      // branch) and just needs a real judge pass rather than a full
      // planner/investigator/implementer redo.
      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(pipelineId, wtDir, workBranch, entry.id, pipelineSession, entry.judgeOnly);
        pipelineSession.taskResults.push(result);
        return result;
      });

      // Merge + push sequentially -- ref updates on the shared repo, one at a
      // time, in the declared task order for this phase.
      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]);
        broadcast("event", 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"]]) --
    // caller's responsibility to supply real phase grouping (see tasks/INDEX.md;
    // filename/numeric sort does NOT match execution order on boards like this).
    // A flat array of ids is also accepted and treated as one single phase.
    // Phases run strictly sequentially (a phase boundary is a real dependency
    // gate); tasks within a phase run concurrently, each in its own worktree --
    // see runPhase.
    function runPipeline({ pipelineId, repo, baseBranch, tasks, branchName }) {
      const cwd = path.join(WORK_DIR, pipelineId);
      fs.mkdirSync(cwd, { recursive: true });

      const phases = Array.isArray(tasks[0]) ? tasks : [tasks];

      const pipelineSession = startSession("pipeline", {
        id: pipelineId,
        pipelineId,
        stage: "pipeline",
        taskResults: [],
        activeTasks: {},
        totalTasks: phases.flat().length,
      });

      (async () => {
        const clone = await runStageWithResolver(
          pipelineId,
          cwd,
          "planner",
          `Run exactly this command, verbatim, no variation: git clone --branch ${baseBranch} ${repo} . -- the trailing dot is required, it clones directly into the current directory instead of creating a subdirectory. Do not cd anywhere first or after. Do nothing else.`,
          "clone"
        );
        if (clone.code !== 0) return endSession(pipelineSession, "clone-crashed");
        if (!fs.existsSync(path.join(cwd, ".git"))) {
          // The model deciding to `cd` elsewhere before cloning (instead of
          // cloning into the assigned cwd) is a real failure mode seen in
          // practice, not a hypothetical -- exit code 0 doesn't mean the clone
          // landed where every later stage's cwd assumes it did.
          return endSession(pipelineSession, "clone-missing");
        }

        // Dedicated branch, never main -- and pushed after every single task
        // (not just at the end) so a pod restart mid-run loses at most the
        // in-progress task's work, not everything since the start.
        const workBranch = branchName || `agent-run/${pipelineId}`;
        const branchResult = await runGit(cwd, ["checkout", "-b", workBranch]);
        if (branchResult.code !== 0) {
          pipelineSession.gitError = branchResult.out;
          return endSession(pipelineSession, "branch-crashed");
        }

        // Seen in practice: a task manually downloads a dependency tarball
        // (crates.io registry access isn't guaranteed from every sandboxed
        // checkout) and it lands at repo root, outside whatever .gitignore
        // already covers -- then git add -A (ours or the model's own) commits
        // it. Append broad build-artifact/archive patterns before any task
        // runs, so it's excluded regardless of who stages files later.
        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 completion sentinel files, harness bookkeeping only",
          ".task-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"]);

        for (const phaseTasks of phases) {
          await runPhase(pipelineId, cwd, workBranch, phaseTasks, pipelineSession);
        }

        const crashed = pipelineSession.taskResults.filter((r) => r.status.endsWith("-crashed"));
        endSession(pipelineSession, crashed.length > 0 ? "completed-with-crashes" : "completed");
      })();

      return pipelineSession;
    }

    const server = http.createServer((req, res) => {
      const url = new URL(req.url, "http://localhost");

      if (url.pathname === "/healthz") {
        res.writeHead(200).end();
        return;
      }

      if (url.pathname === "/sessions" && req.method === "GET") {
        res.writeHead(200, { "Content-Type": "application/json" });
        res.end(JSON.stringify([...sessions.values()]));
        return;
      }

      if (url.pathname === "/run" && req.method === "POST") {
        let body = "";
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => {
          try {
            const { agent, prompt, provider, model } = JSON.parse(body);
            if (!agent || !prompt) throw new Error("agent and prompt are required");
            const session = runAgent(agent, prompt, { provider, model });
            res.writeHead(200, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ id: session.id }));
          } catch (err) {
            res.writeHead(400, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ error: err.message }));
          }
        });
        return;
      }

      if (url.pathname === "/pipeline" && req.method === "POST") {
        let body = "";
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => {
          try {
            const { repo, baseBranch, tasks, branchName } = JSON.parse(body);
            if (!repo || !baseBranch || !Array.isArray(tasks) || tasks.length === 0) {
              throw new Error("repo, baseBranch, and a non-empty tasks array are required");
            }
            const pipelineId = crypto.randomUUID();
            runPipeline({ pipelineId, repo, baseBranch, tasks, branchName });
            res.writeHead(200, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ id: pipelineId }));
          } catch (err) {
            res.writeHead(400, { "Content-Type": "application/json" });
            res.end(JSON.stringify({ error: err.message }));
          }
        });
        return;
      }

      res.writeHead(404).end();
    });

    const wss = new WebSocketServer({ server, path: "/console" });
    wss.on("connection", (ws) => {
      for (const session of sessions.values()) {
        ws.send(JSON.stringify({ type: "snapshot", session }));
      }
      viewers.add(ws);
      ws.on("close", () => viewers.delete(ws));
    });

    server.listen(PORT, () => console.log(`agent-hub listening on :${PORT}`));
kind: ConfigMap
metadata:
  name: hub-src
  namespace: agent-pod