From 35bef19e0c1acd8e66afb7e4134ef8edde2b5635 Mon Sep 17 00:00:00 2001 From: Story Crater Bot <19826264+Riotpiaole@users.noreply.github.com> Date: Tue, 18 Aug 2026 21:30:49 -0700 Subject: [PATCH] feat(agent-pod): persistent per-role agent pool, concurrency moves to repo level coordinator.js now runs one long-lived planner/investigator/implementer/judge session per repo (reused across every task via tmux send-keys) instead of a fresh spawn per task per stage. Tasks within a repo run sequentially against that pool; concurrency is now REPO_CONCURRENCY (default 3) concurrent repos via a new --repos flag, not concurrent tasks in one repo's phase. --- k8s/apps/agent-pod/coordinator-configmap.yaml | 543 +++++++----------- k8s/apps/agent-pod/deployment.yaml | 20 +- 2 files changed, 221 insertions(+), 342 deletions(-) diff --git a/k8s/apps/agent-pod/coordinator-configmap.yaml b/k8s/apps/agent-pod/coordinator-configmap.yaml index 2e99b6a..92552f3 100644 --- a/k8s/apps/agent-pod/coordinator-configmap.yaml +++ b/k8s/apps/agent-pod/coordinator-configmap.yaml @@ -8,20 +8,35 @@ data: #!/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 + // hub.js's old /pipeline handler. Every interactive stage runs through the + // patched agent-manager fork's `spawn` subcommand, so it is a real tracked + // session (tmux pane on agent-manager's private socket + a state.db row) + // from the moment it exists -- attachable and visible in agent-manager's + // own TUI the whole time it runs. + // + // Per repo, each role (planner/investigator/implementer/judge) is ONE + // persistent agent-manager session, not a fresh spawn per task: the first + // task to need a role spawns it, every later task for that role reuses the + // same tmux pane via `tmux send-keys` (see runOnPool) -- the same nudge + // mechanism that used to only fire on a stall now doubles as "give this + // agent its next task." A role's conversation history accumulates across + // every task it ever handles for that repo; nothing resets it mid-pipeline. + // Because only one implementer/judge/etc. exists per repo, tasks within a + // phase run strictly sequentially against the pool -- no per-task worktree, + // no per-task branch, no merge-back step; every task commits straight onto + // the phase branch in the repo's one shared clone. + // + // The unit of concurrency is now the REPO, not the task: runCoordinator + // takes a list of repos and runs up to REPO_CONCURRENCY of them at once, + // each with its own clone (under WORK_DIR/) and its own 4-agent + // pool. The coordinator never kills a role's session; it rests at an idle + // prompt between tasks, and agent-manager's session list becomes the audit + // trail of everything every repo's pipeline ran. Completion is signaled by + // sentinel files under the repo's clone (unchanged convention), waited on // with fs.watch instead of polling. const fs = require("node:fs"); const path = require("node:path"); + const crypto = require("node:crypto"); const { spawn } = require("node:child_process"); const WORK_DIR = process.env.HUB_WORK_DIR || path.join(require("node:os").tmpdir(), "agent-harness-work"); @@ -37,31 +52,29 @@ data: // 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) -- + // route every stage -- headless (spawnPi) and interactive (runOnPool) -- // at the homelab model instead, e.g. AGENT_PROVIDER=homelab-ornith // AGENT_MODEL=ornith:35b. const AGENT_PROVIDER = process.env.AGENT_PROVIDER || ""; const AGENT_MODEL = process.env.AGENT_MODEL || ""; - // judge (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. + // judge can run a different model than the rest of the chain, e.g. + // homelab-reasoning instead of homelab-ornith now that verifier/PRM is + // retired. Falls back to AGENT_PROVIDER/AGENT_MODEL when unset, so a run + // that doesn't care keeps one uniform model everywhere. const JUDGE_PROVIDER = process.env.JUDGE_PROVIDER || AGENT_PROVIDER; const JUDGE_MODEL = process.env.JUDGE_MODEL || AGENT_MODEL; - function providerModelFor(stageLabel) { - return stageLabel === "judge" ? { provider: JUDGE_PROVIDER, model: JUDGE_MODEL } : { provider: AGENT_PROVIDER, model: AGENT_MODEL }; + function providerModelFor(role) { + return role === "judge" ? { provider: JUDGE_PROVIDER, model: JUDGE_MODEL } : { provider: AGENT_PROVIDER, model: AGENT_MODEL }; } // agent-manager's private tmux server and session-naming scheme // (internal/tmux/tmux.go: defaultSocket = "agentmgr", sessionName(id) = // "am_"+id) -- stable, documented internals of the fork, used here only - // for read-only introspection (pane capture) and 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. + // for read-only introspection (pane capture) and role nudges, exactly the + // class of operation hub.js already ran directly against its own sessions + // rather than asking a model to do it. const AM_SOCKET = "agentmgr"; function amSessionName(id) { return `am_${id}`; @@ -112,8 +125,8 @@ data: // Headless one-shot pi call (`pi -p --mode json `), 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. + // exactly as before -- only the interactive per-role stages (runOnPool, + // below) go through agent-manager. function spawnPi({ agent, prompt, cwd }) { const finalPrompt = ROLE_SKILLS.has(agent) ? `/skill:${agent} ${HARD_RULES}${prompt}` : prompt; const args = ["-p", "--mode", "json"]; @@ -154,99 +167,17 @@ data: }); } - async function askResolver(cwd, task, diagnosticPrompt) { + async function askResolver(cwd, repoId, 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. + // after limitMs with no sign of it -- event-driven completion instead of + // hub.js's old 10s poll. function waitForSentinel(filePath, limitMs) { return new Promise((resolve) => { if (fs.existsSync(filePath)) return resolve(true); @@ -282,119 +213,104 @@ data: }); } - // 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}`; + // Runs one task's worth of work on a persistent per-role agent: spawns the + // role's session the first time it's ever needed for this repo, sends every + // later prompt into that same tmux pane via send-keys. pool is a plain + // object keyed by role name ("planner"/"investigator"/"implementer"/ + // "judge"), shared across every task in a repo's pipeline (see + // runRepoPipeline) -- it IS the 4-agent pool, one entry per role, filled in + // lazily as each role gets its first task. + async function runOnPool(pool, cwd, repoId, role, prompt, sentinelFile) { fs.rmSync(sentinelFile, { force: true }); + const label = `${repoId}-${role}`; + let target = pool[role]; - 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); + if (!target) { + const spawnArgs = ["spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--prompt", HARD_RULES + prompt]; + const { provider, model } = providerModelFor(role); + if (provider) spawnArgs.push("--provider", provider); + if (model) spawnArgs.push("--model", model); + const spawned = await runCmd(AGENT_MANAGER_BIN, spawnArgs); + if (spawned.code !== 0) return { ok: false, crashed: true, error: spawned.out, sessionName: label }; + target = amSessionName(spawned.out); + pool[role] = target; + } else { + await runAmTmux(["send-keys", "-t", target, HARD_RULES + prompt, "Enter"]); + } let ok = await waitForSentinel(sentinelFile, STAGE_TIMEOUT_MS); if (!ok) { const pane = await runAmTmux(["capture-pane", "-t", target, "-p", "-S", "-200"]); const resolution = await askResolver( cwd, - task, - `Task ${task}'s "${stageLabel}" stage hasn't finished after 10 minutes. Its pane tail:\n${pane.out.slice(-3000)}\n\n` + + repoId, + `Repo ${repoId}'s "${role}" agent hasn't finished its current task after 10 minutes. Its pane tail:\n${pane.out.slice(-3000)}\n\n` + `Decide: is it still making real progress and worth nudging to wrap up, or stuck and worth abandoning?` ); if (resolution === "RETRY") { - await runAmTmux([ - "send-keys", - "-t", - target, - `Please wrap up 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") { + await runAmTmux(["send-keys", "-t", target, `Please wrap up now and touch ${path.basename(sentinelFile)} when done.`, "Enter"]); ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS); } } return { ok, sessionName: label }; } + function plannerPrompt(task, specHint) { + return ( + `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` + ); + } + + function investigatorPrompt(task) { + return ( + `Use the investigator skill to confirm PLAN.md against real sources for task ${task}, append findings, commit. ` + + `Then run: touch .stage-done-${task}-investigator` + ); + } + + function implementerPrompt(task, attempt, feedbackHint) { + return ( + `Use the implementer skill to implement what the current PLAN.md specifies for task ${task} (commit as you go). ` + + `${feedbackHint} Then run: touch .stage-done-${task}-implementer-${attempt}` + ); + } + + function judgePrompt(task, baseBranch, attempt) { + return ( + `Use the judge skill to review the diff against ${baseBranch}...HEAD for task ${task}. 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}` + ); + } + const MAX_IMPLEMENT_ATTEMPTS = 5; const MAX_PLAN_REVISIONS = 3; - // Runs one task in its own git worktree (see runPhase): planner drafts + // Runs one task against the repo's shared role pool: planner drafts // PLAN.md, investigator confirms it, then implementer and judge go back and // forth -- judge's FAIL rationale lands in .task-result-, 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) { + // MAX_IMPLEMENT_ATTEMPTS straight fails, the SAME planner agent is asked to + // judge whether the plan itself is wrong (it still remembers drafting it); + // if so it revises PLAN.md and the implementer gets a fresh attempt budget. + // MAX_PLAN_REVISIONS caps this from looping forever on a task that's + // genuinely stuck. All work happens directly in cwd (the repo's one shared + // clone, currently checked out to the phase branch) -- no worktree, since + // only one implementer/judge exist per repo and tasks run strictly one at a + // time (see runPhase). + async function runTaskOnPool(cwd, baseBranch, task, pool, repoId, pipelineSession, judgeOnly) { const resultFile = path.join(cwd, `.task-result-${task}`); fs.rmSync(resultFile, { force: true }); const specHint = `the file under tasks/ starting with "${task}-"`; - // 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() }; + const stage = async (role, prompt, sentinel, displayLabel) => { + const label = displayLabel || role; + pipelineSession.activeTasks[task] = { stage: label, startedAt: new Date().toISOString() }; logProgress(pipelineSession); - const result = await 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})`); + const result = await runOnPool(pool, cwd, repoId, role, prompt, sentinel); + await commitPending(cwd, `task: ${task} (${label})`); return result; }; @@ -411,7 +327,7 @@ data: }; if (judgeOnly) { - const quick = await spawnRole( + const quick = await stage( "judge", `Task ${task} may already be implemented on this branch -- check ` + `\`git log --oneline --grep '${task}'\` and the current code directly against its spec ` + @@ -430,16 +346,10 @@ data: } } - // 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`)); + let result = await stage("planner", plannerPrompt(task, 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`)); + result = await stage("investigator", investigatorPrompt(task), path.join(cwd, `.stage-done-${task}-investigator`)); if (!result.ok) return abandon("investigator", result); let planRevisions = 0; @@ -447,13 +357,6 @@ data: 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++; @@ -465,22 +368,14 @@ data: : ""; 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; + result = await stage( + "implementer", + implementerPrompt(task, implementAttempt, feedbackHint), + path.join(cwd, `.stage-done-${task}-implementer-${implementAttempt}`) + ); 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}`)); + result = await stage("judge", judgePrompt(task, baseBranch, 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") : ""; @@ -490,15 +385,16 @@ data: if (implementAttempt >= MAX_IMPLEMENT_ATTEMPTS) { if (planRevisions >= MAX_PLAN_REVISIONS) break; planRevisions++; - result = await spawnRole( - "planner-revise", + result = await stage( + "planner", `Implementer failed judge review ${MAX_IMPLEMENT_ATTEMPTS} times in a row for task ${task}. Read PLAN.md, ` + `the judge's feedback in .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}`) + path.join(cwd, `.stage-done-${task}-planner-revise-${planRevisions}`), + "planner-revise" ); if (!result.ok) return abandon("planner-revise", result, planRevisions); implementAttempt = 0; @@ -515,73 +411,17 @@ data: 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) { + // Runs every task in a phase (no declared dependency between them) strictly + // one at a time against the repo's shared role pool -- only one implementer/ + // judge/etc. exists per repo, so there is no per-task concurrency to have + // here anymore (see REPO_CONCURRENCY below for where concurrency now + // lives). No worktrees: every task commits directly onto phaseBranch in the + // one shared cwd. + async function runPhase(cwd, baseBranch, phaseTasks, pool, repoId, pipelineSession) { const entries = phaseTasks.map((t) => (typeof t === "string" ? { id: t, judgeOnly: false } : t)); - - 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); + const result = await runTaskOnPool(cwd, baseBranch, entry.id, pool, repoId, 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); } } @@ -625,31 +465,27 @@ data: } function logProgress(pipelineSession) { - console.log(`[pipeline ${pipelineSession.id}] ${JSON.stringify(pipelineSession)}`); + console.log(`[repo ${pipelineSession.id}] ${JSON.stringify(pipelineSession)}`); } + // Runs one repo's full pipeline: clone, then phases strictly sequentially. // tasks: array of phases, each phase an array of task ids with no declared // dependency on each other (e.g. [["T0.1","T0.2"], ["T1.1","T1.2","T1.3"]]). - // A flat array of ids is also accepted and treated as one single phase. - // 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//, 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); + // A flat array of ids is also accepted and treated as one single phase. If + // omitted, phases are discovered from the repo's own tasks/INDEX.md and run + // judgeOnly first (a cheap "is this already done" check against the + // board's possibly-stale checkmarks). Each phase gets its own branch + // (agent-run//, e.g. .../T1); once every task in that + // phase lands "done" or "done-with-concerns" AND the phase judge (the same + // pooled judge agent that reviewed each task) passes the integration + // review, the phase branch is squash-merged into baseBranch and pushed, + // then the next phase branches off that updated base. Any failure halts + // this repo's pipeline before merging -- it does not affect other repos + // running concurrently (see runCoordinator). + async function runRepoPipeline({ repoId, repo, baseBranch, tasks, branchName }, pipelineSession) { + const cwd = path.join(WORK_DIR, repoId); fs.mkdirSync(cwd, { recursive: true }); - - const pipelineSession = { - id: pipelineId, - status: "running", - taskResults: [], - activeTasks: {}, - totalTasks: 0, - startedAt: new Date().toISOString(), - }; + const pool = {}; const finish = (status) => { pipelineSession.status = status; @@ -670,15 +506,9 @@ data: } 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"; + pipelineSession.gitError = "no tasks given and tasks/INDEX.md not found or empty"; return finish("no-tasks-found"); } if (!tasks) { @@ -689,7 +519,7 @@ data: 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 phaseBranch = branchName ? `${branchName}/${phaseLabel}` : `agent-run/${repoId}/${phaseLabel}`; const branchResult = await runGit(cwd, ["checkout", "-b", phaseBranch]); if (branchResult.code !== 0) { @@ -720,7 +550,7 @@ data: await runGit(cwd, ["commit", "-m", "chore: broaden .gitignore for agent-run artifacts"]); } - await runPhase(pipelineId, cwd, phaseBranch, phaseTasks, pipelineSession); + await runPhase(cwd, phaseBranch, phaseTasks, pool, repoId, pipelineSession); const phaseTaskIds = new Set(phaseTasks.map((t) => (typeof t === "string" ? t : t.id))); const phaseResults = pipelineSession.taskResults.filter((r) => phaseTaskIds.has(r.task)); @@ -734,20 +564,19 @@ data: 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( + const phaseJudge = await runOnPool( + pool, cwd, - `phase-${phaseLabel}`, + repoId, "judge", - HARD_RULES + - `Use the judge skill to review the full phase diff for phase ${phaseLabel} against ` + + `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 + path.join(cwd, `.stage-done-phase-${phaseLabel}-judge`) ); await commitPending(cwd, `phase: ${phaseLabel} integration review`); @@ -791,11 +620,57 @@ data: return finish("completed"); } + async function runConcurrent(items, limit, worker) { + const results = new Array(items.length); + let i = 0; + async function next() { + while (i < items.length) { + const idx = i++; + results[idx] = await worker(items[idx], idx); + } + } + await Promise.all(Array.from({ length: Math.min(limit, items.length) }, next)); + return results; + } + + // How many repos can be mid-flight at once. Each repo gets its own clone + // and its own 4-agent pool (planner/investigator/implementer/judge), so + // this is now the real concurrency knob -- tasks within one repo are + // already serialized against that repo's pool (see runPhase). Backends + // that queue rather than reject concurrent requests to one loaded model + // (e.g. Ollama) still process this many repos' worth of stages one at a + // time internally even if REPO_CONCURRENCY says otherwise; bump it once + // there's more than one model instance to actually parallelize against. + const REPO_CONCURRENCY = Number(process.env.REPO_CONCURRENCY) || 3; + + // Top-level entry point: runs every repo in `repos` to completion, up to + // REPO_CONCURRENCY at a time. Returns a map of repoId -> final + // pipelineSession, one per repo, independent of how the others fared. + async function runCoordinator({ repos, base, tasks, branchName }) { + const sessions = {}; + await runConcurrent(repos, REPO_CONCURRENCY, async (repoUrl) => { + const repoId = crypto.randomUUID(); + const pipelineSession = { + id: repoId, + repo: repoUrl, + status: "running", + taskResults: [], + activeTasks: {}, + totalTasks: 0, + startedAt: new Date().toISOString(), + }; + sessions[repoId] = pipelineSession; + await runRepoPipeline({ repoId, repo: repoUrl, baseBranch: base, tasks, branchName }, pipelineSession); + }); + return sessions; + } + function parseArgs(argv) { const opts = { base: "main" }; for (let i = 0; i < argv.length; i++) { const a = argv[i]; if (a === "--repo") opts.repo = argv[++i]; + else if (a === "--repos") opts.repos = argv[++i]; else if (a === "--base") opts.base = argv[++i]; else if (a === "--tasks") opts.tasks = argv[++i]; else if (a === "--branch") opts.branch = argv[++i]; @@ -805,10 +680,13 @@ data: async function main() { const opts = parseArgs(process.argv.slice(2)); - if (!opts.repo) { + const repos = opts.repos ? opts.repos.split(",") : opts.repo ? [opts.repo] : null; + if (!repos || repos.length === 0) { console.error( - "usage: coordinator.js --repo [--tasks T0.1,T0.2;T1.1,T1.2,...] [--base main] [--branch ]\n" + - " --tasks omitted: discovers phases from the cloned repo's own tasks/INDEX.md" + "usage: coordinator.js --repos [--tasks T0.1,T0.2;T1.1,T1.2,...] [--base main] [--branch ]\n" + + " --repo also accepted for a single repo\n" + + " --tasks applies to every repo listed; omitted: each repo discovers its own phases from tasks/INDEX.md\n" + + " REPO_CONCURRENCY env var (default 3): how many repos run at once" ); // process.exitCode + natural exit, not process.exit() -- stdout piped // through kubectl exec (not a TTY) can drop buffered console.log/ @@ -819,13 +697,12 @@ data: 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; + const sessions = await runCoordinator({ repos, base: opts.base, tasks: phases, branchName: opts.branch }); + process.exitCode = Object.values(sessions).every((s) => s.status === "completed") ? 0 : 1; } if (require.main === module) { main(); } - module.exports = { runPipeline, spawnCommandLine, spawnAndWaitStage, waitForStage, parseTaskBoard }; + module.exports = { runCoordinator, runRepoPipeline, runOnPool, parseTaskBoard }; diff --git a/k8s/apps/agent-pod/deployment.yaml b/k8s/apps/agent-pod/deployment.yaml index be498a7..13a65e2 100644 --- a/k8s/apps/agent-pod/deployment.yaml +++ b/k8s/apps/agent-pod/deployment.yaml @@ -31,15 +31,17 @@ spec: # Also builds the agent-manager fork (github.com/Riotpiaole/ # agent-manager, add-headless-spawn branch) from source and drops # coordinator.js in beside hub.js -- neither is the container's - # foreground process. hub.js keeps that role unchanged; running - # multiple repos' pipelines concurrently in this one pod means - # `kubectl exec -- node /root/coordinator.js --repo X --tasks - # Y &` once per repo, each an independent process inside the same - # container, each spawning its own agent-manager-tracked sessions - # on the container's local tmux server -- `kubectl exec -it - # -- agent-manager` attaches its TUI live against those same - # sessions, no cross-machine visibility problem since spawner, - # tmux server, and viewer are all colocated here. + # foreground process. hub.js keeps that role unchanged; coordinator.js + # itself now owns multi-repo concurrency (REPO_CONCURRENCY env, + # default 3), so one invocation handles every repo: + # `kubectl exec -- node /root/coordinator.js --repos + # repoA,repoB,... --tasks ...`. Each repo gets its own clone and its + # own persistent 4-agent pool (planner/investigator/implementer/ + # judge, one agent-manager session per role, reused across every + # task in that repo) on the container's local tmux server -- + # `kubectl exec -it -- agent-manager` attaches its TUI live + # against those same sessions, no cross-machine visibility problem + # since spawner, tmux server, and viewer are all colocated here. # # No prebuilt Linux binary is shipped for agent-manager: the local # .bin/ build is macOS arm64 (wrong OS/arch for this container