rock/homelab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

coordinator: detect+respawn dead pool sessions, bound resolver call

Browse Source 
Dead sessions were only caught after a full 10-min stall timeout; now
polled via agent-manager status and respawned (retry once). spawnPi had
no timeout and could hang a repo's whole pipeline forever -- bounded to
5 minutes now.
This commit is contained in:
Story Crater Bot
2026-08-19 20:27:17 -07:00
parent 052f235d7b
commit 1b38297cbf
1 changed files with 106 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
k8s/apps/agent-pod/coordinator-configmap.yaml
+106 -18
View File
@@ -52,6 +52,14 @@ data:
const AGENT_MANAGER_BIN = process.env.AGENT_MANAGER_BIN || path.join(__dirname, "..", ".bin", "agent-manager-fork");
// agent-manager's own session-state DB -- used to detect a session that has
// actually died (process crashed/exited, status flips to "errored"/"dead")
// instead of one that's merely slow. Read-only introspection plus the one
// UPDATE in killDeadSession below, same class of operation as the tmux
// nudges already done directly against agent-manager's internals.
const AGENT_MANAGER_DB =
process.env.AGENT_MANAGER_DB || path.join(require("node:os").homedir(), ".config", "agent-manager", "state.db");
// Empty means "let pi fall back to ~/.pi/agent/settings.json's default"
// (currently anthropic/claude-sonnet-4-5, real paid usage). Set both to
// route every stage -- headless (spawnPi) and interactive (runOnPool) --
@@ -161,6 +169,17 @@ data:
// session: resolver's crash/stall diagnosis, and the initial clone. Kept
// exactly as before -- only the interactive per-role stages (runOnPool,
// below) go through agent-manager.
//
// Bounded by SPAWN_PI_TIMEOUT_MS -- unlike runOnPool's pooled sessions
// (which now have status polling to catch a dead session fast, see
// waitForSentinel/killDeadSession), this is a raw child_process with no
// equivalent escape hatch. Observed live: a resolver call shared the
// default backend with a concurrently-busy repo's implementer and sat for
// 6+ minutes producing nothing -- with no timeout here, that blocks the
// entire calling repo's pipeline forever, since askResolver is always
// awaited before the next stage can run.
const SPAWN_PI_TIMEOUT_MS = 5 * 60 * 1000;
function spawnPi({ agent, prompt, cwd }) {
const finalPrompt = ROLE_SKILLS.has(agent) ? `/skill:${agent} ${HARD_RULES}${prompt}` : prompt;
const args = ["-p", "--mode", "json"];
@@ -197,7 +216,19 @@ data:
stderrTail = (stderrTail + chunk.toString()).slice(-4000);
});
return new Promise((resolve) => {
child.on("close", (code) => resolve({ code, lastText, stderrTail }));
let settled = false;
const timer = setTimeout(() => {
if (settled) return;
settled = true;
child.kill("SIGKILL");
resolve({ code: null, lastText, stderrTail, timedOut: true });
}, SPAWN_PI_TIMEOUT_MS);
child.on("close", (code) => {
if (settled) return;
settled = true;
clearTimeout(timer);
resolve({ code, lastText, stderrTail });
});
});
}
@@ -209,19 +240,27 @@ data:
const STAGE_TIMEOUT_MS = 10 * 60 * 1000;
const NUDGE_TIMEOUT_MS = 5 * 60 * 1000;
// Resolves as soon as filePath appears (fs.watch on its directory), or
// after limitMs with no sign of it -- event-driven completion instead of
// hub.js's old 10s poll.
function waitForSentinel(filePath, limitMs) {
// Resolves as soon as filePath appears (fs.watch on its directory, same as
// before), as soon as target's agent-manager status flips to "errored" or
// "dead" (polled -- state.db has no watch mechanism), or after limitMs with
// neither. A session that has actually crashed will never touch the
// sentinel, so without the status poll this just burns the full STAGE_
// TIMEOUT_MS waiting on a file that was never coming, same as a genuine
// stall -- polling status catches that in ~pollMs instead.
function waitForSentinel(filePath, target, limitMs, pollMs = 5000) {
return new Promise((resolve) => {
if (fs.existsSync(filePath)) return resolve(true);
if (fs.existsSync(filePath)) return resolve({ ok: true });
const dir = path.dirname(filePath);
const id = target.replace(/^am_/, "");
let settled = false;
let watcher;
let poller;
let timer;
const finish = (result) => {
if (settled) return;
settled = true;
clearTimeout(timer);
clearInterval(poller);
if (watcher) {
try {
watcher.close();
@@ -233,20 +272,36 @@ data:
};
try {
watcher = fs.watch(dir, () => {
if (fs.existsSync(filePath)) finish(true);
if (fs.existsSync(filePath)) finish({ ok: true });
});
} catch {
// dir missing at watch time is a real bug elsewhere (cwd should
// already exist); surface it as a timeout rather than hang forever.
return finish(false);
return finish({ timedOut: true });
}
// Closes the race between the existsSync check above and the watcher
// actually being attached.
if (fs.existsSync(filePath)) return finish(true);
const timer = setTimeout(() => finish(false), limitMs);
if (fs.existsSync(filePath)) return finish({ ok: true });
poller = setInterval(async () => {
const { out } = await runCmd("sqlite3", [AGENT_MANAGER_DB, `SELECT status FROM sessions WHERE id='${id}'`]);
const status = out.trim();
if (status === "errored" || status === "dead") finish({ dead: true, status });
}, pollMs);
timer = setTimeout(() => finish({ timedOut: true }), limitMs);
});
}
// Kills a session that's actually crashed (not just slow) and archives it
// in agent-manager's own DB so it stops showing up as a live, unattended
// pane -- otherwise every crash leaves an orphaned tmux session + state.db
// row behind permanently, identical to the manually-cleaned-up poiman-
// planner ghost session found earlier this same run.
async function killDeadSession(target) {
await runAmTmux(["kill-session", "-t", target]);
const id = target.replace(/^am_/, "");
await runCmd("sqlite3", [AGENT_MANAGER_DB, `UPDATE sessions SET archived=1 WHERE id='${id}'`]);
}
// Runs one task's worth of work on a persistent per-role agent: spawns the
// role's session the first time it's ever needed for this repo, sends every
// later prompt into that same tmux pane via send-keys -- prefixed with a
@@ -257,10 +312,14 @@ data:
// 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.
// A dead/errored session gets one respawn-and-retry (same prompt, fresh
// session) before this stage is abandoned -- matches resolver-SKILL.md's
// own documented contract of retrying a failed stage at most once.
const DEAD_SESSION_RETRIES = 1;
async function runOnPool(pool, cwd, repoId, role, prompt, sentinelFile) {
fs.rmSync(sentinelFile, { force: true });
const label = `${repoId}-${role}`;
let target = pool[role];
// A pooled session's shell cwd drifts as it explores the repo (e.g. cd
// into a Rust workspace subdirectory to read source) and nothing resets
@@ -275,14 +334,19 @@ data:
// wrong by reasoning about a relative "current directory."
const cwdReminder = `Your working directory for this task is ${cwd} -- if your shell isn't already there, run: cd ${cwd}\n\n`;
if (!target) {
const spawnFresh = async () => {
const spawnArgs = ["spawn", "--tool", "pi", "--cwd", cwd, "--name", label, "--group", repoId, "--prompt", cwdReminder + HARD_RULES + prompt];
const { provider, model } = providerModelFor(role);
if (provider) spawnArgs.push("--provider", provider);
if (model) spawnArgs.push("--model", model);
const spawned = await runCmd(AGENT_MANAGER_BIN, spawnArgs);
if (spawned.code !== 0) return { ok: false, crashed: true, error: spawned.out, sessionName: label };
target = amSessionName(spawned.out);
return spawned.code === 0 ? amSessionName(spawned.out) : null;
};
let target = pool[role];
if (!target) {
target = await spawnFresh();
if (!target) return { ok: false, crashed: true, error: "spawn failed", sessionName: label };
pool[role] = target;
} else {
await runAmTmux(["send-keys", "-t", target, "/new", "Enter"]);
@@ -290,8 +354,25 @@ data:
await runAmTmux(["send-keys", "-t", target, cwdReminder + HARD_RULES + prompt, "Enter"]);
}
let ok = await waitForSentinel(sentinelFile, STAGE_TIMEOUT_MS);
if (!ok) {
for (let deadRetries = 0; ; deadRetries++) {
const outcome = await waitForSentinel(sentinelFile, target, STAGE_TIMEOUT_MS);
if (outcome.ok) return { ok: true, sessionName: label };
if (outcome.dead) {
await killDeadSession(target);
if (pool[role] === target) delete pool[role];
if (deadRetries >= DEAD_SESSION_RETRIES) {
return { ok: false, crashed: true, error: `session died (status: ${outcome.status})`, sessionName: label };
}
target = await spawnFresh();
if (!target) return { ok: false, crashed: true, error: "respawn after death failed", sessionName: label };
pool[role] = target;
continue;
}
// Plain stall -- session still alive, just slow. Ask resolver once,
// nudge if it says worth it, and stop here either way (this is not
// the death path, so no respawn/retry loop).
const pane = await runAmTmux(["capture-pane", "-t", target, "-p", "-S", "-200"]);
const resolution = await askResolver(
cwd,
@@ -299,12 +380,18 @@ data:
`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?`
);
let ok = false;
if (resolution === "RETRY") {
await runAmTmux(["send-keys", "-t", target, `Please wrap up now and run: touch ${sentinelFile}`, "Enter"]);
ok = await waitForSentinel(sentinelFile, NUDGE_TIMEOUT_MS);
const nudged = await waitForSentinel(sentinelFile, target, NUDGE_TIMEOUT_MS);
ok = nudged.ok === true;
if (nudged.dead) {
await killDeadSession(target);
if (pool[role] === target) delete pool[role];
}
}
return { ok, sessionName: label };
}
return { ok, sessionName: label };
}
function plannerPrompt(task, specHint, judgeOnly, cwd) {
@@ -876,3 +963,4 @@ data:
}
module.exports = { runCoordinator, runRepoPipeline, runOnPool, parseTaskBoard };