701 lines
30 KiB
YAML
701 lines
30 KiB
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
|