fix(coding-agent): reclaim stale session state, bound worker recovery, and heal via doctor
1- Fixed stale session lockups by reclaiming dead worker leases, bounding recovery retries, and repairing orphaned state via doctor --fix.1234567891011121314101516171819133 unmodified linesExpand all153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191import { createHash } from "node:crypto";import { existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";import { tmpdir } from "node:os";import { join } from "node:path";import { describe, expect, it } from "vitest";import { type DaemonInfo, evaluateShutdownQuietPeriod, planReap, planShutdownAll, planShutdownConfirmation, reapStaleWorkerSessions, verifyHelloSupervisorPid,} from "../src/cli/daemon-ps.js";import { getProcessStartId } from "../src/core/session-lease.js";import { canonicalSessionPath, getProcessStartId } from "../src/core/session-lease.js";
describe("evaluateShutdownQuietPeriod", () => { it("requires a full quiet period independently of the convergence window", () => { expect(evaluateShutdownQuietPeriod(10_500, 10_000)).toBe("waiting");133 unmodified linesExpand all isDefault: false, ...options, };}
describe("reapStaleWorkerSessions", () => { it("detects and clears recovering worker with dead pid, stale lease, and socket", () => { const root = mkdtempSync(join(tmpdir(), "prime-reap-test-")); try { const sessionFile = join(root, "session-test.jsonl"); const socketPath = join(root, "worker.sock"); const workerDir = join(root, "daemon-workers", "test-hash"); mkdirSync(workerDir, { recursive: true }); const descPath = join(workerDir, "w1.json"); const journalPath = join(workerDir, "w1.recovery.jsonl"); writeFileSync( descPath, `{"version":2,"workerId":"w1","pid":2147483647,"socketPath":"${socketPath}","recoveryJournalPath":"${journalPath}","supervisorSocketPath":"/tmp/s.sock","sessionFile":"${sessionFile}","lifecycle":"recovering","consecutiveFailures":8}`, ); writeFileSync(journalPath, "{}\n"); writeFileSync(socketPath, ""); const leaseKey = createHash("sha256").update(canonicalSessionPath(sessionFile)).digest("hex"); const leaseDir = join(root, "session-leases", `${leaseKey}.lock`); mkdirSync(leaseDir, { recursive: true }); writeFileSync(join(leaseDir, "owner.json"), `{"version":1,"pid":2147483647,"sessionPath":"${sessionFile}"}`);
const reaped = reapStaleWorkerSessions(root); expect(reaped).toEqual([ expect.objectContaining({ socketPath: "session session-test", action: expect.stringContaining("cleared") }), ]); expect(existsSync(descPath)).toBe(false); expect(existsSync(journalPath)).toBe(false); expect(existsSync(leaseDir)).toBe(false); expect(existsSync(socketPath)).toBe(false); } finally { rmSync(root, { recursive: true, force: true }); } });});4633 unmodified linesExpand all46344635463646374638463946404641464246434644464546464647464846494650465146524653465446554656465746584659466046614662466346644665466646674668466946704671467246734674467546764677467846794680468146824683468446854686468746884689469046914692469346944695469646974698469947004701470247034633 unmodified linesExpand all expect(handshakeBudgetMs(now + 25, now)).toBe(25); expect(() => handshakeBudgetMs(now, now)).toThrow(DaemonWorkerProbeTimeoutError); expect(() => handshakeBudgetMs(now - 1, now)).toThrow(DaemonWorkerProbeTimeoutError); });
it("reclaims dead worker registration instead of failing with session already active", async () => { const worker = { descriptor: { workerId: "dead", pid: 2_147_483_647, rootActiveSessionId: "s", ownerClientId: "o" }, }; const reclaimDeadWorkerRegistration = vi.fn(async () => {}); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { processIdentity: () => "gone", reclaimDeadWorkerRegistration, }) as { assertWorkerCreateOwner: (w: unknown, o: string, p: string) => Promise<void> }; await supervisor.assertWorkerCreateOwner(worker, "curr", "/tmp/s.jsonl"); expect(reclaimDeadWorkerRegistration).toHaveBeenCalledWith(worker); });
it("discards recovering worker record when failure bound is reached", async () => { const worker = { descriptor: { workerId: "ex", pid: 2_147_483_647, rootActiveSessionId: "s", consecutiveFailures: 5 }, }; const stopWorker = vi.fn(async () => {}); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([["ex", worker]]), connectWorker: vi.fn(async () => { throw new Error("fail"); }), assertRecoveryAllowed: vi.fn(async () => {}), persistWorker: vi.fn(), stopWorker, log: vi.fn(), }) as { recoverWorker: (w: unknown) => Promise<void> }; await supervisor.recoverWorker(worker); expect(stopWorker).toHaveBeenCalledWith(worker, true, true); });
it("discards stale worker and creates new one when assigned root is missing", async () => { const worker = { descriptor: { workerId: "m", rootActiveSessionId: "r", lifecycle: "ready" }, client: {} }; const stopWorker = vi.fn(async () => {}); const fresh = { descriptor: { workerId: "fresh" } }; const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([["m", worker]]), workerHasRosterRoot: () => false, assertWorkerCreateOwner: vi.fn(async () => {}), isWorkerReadyForCreate: () => false, canRetryFailedWorker: () => false, stopWorker, launchWorker: vi.fn(async () => fresh), }) as { reuseWorkerForCreate: (w: unknown, o: unknown, p: string, c?: unknown) => Promise<unknown> }; const res = await supervisor.reuseWorkerForCreate(worker, undefined, "/tmp/s.jsonl", { type: "create" }); expect(stopWorker).toHaveBeenCalledWith(worker, true, true); expect(res).toBe(fresh); });
it("reaps worker when adoption hello handshake fails", async () => { const worker = { descriptor: { workerId: "uh", pid: process.pid, rootActiveSessionId: "s" } }; const stopWorker = vi.fn(async () => {}); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { connectWorker: vi.fn(async () => { throw new Error("Timed out waiting for daemon worker hello"); }), assertRecoveryAllowed: vi.fn(async () => {}), stopWorker, log: vi.fn(), }) as { adoptOrRecoverWorker: (w: unknown) => Promise<void> }; await supervisor.adoptOrRecoverWorker(worker); expect(stopWorker).toHaveBeenCalledWith(worker, true, true); });});123452 unmodified lines8910111112131415161000 unmodified linesExpand all101710181019102010201021102110221023102410251026100 unmodified lines11271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203113212041205120612071208168 unmodified linesExpand allimport { createHash } from "node:crypto";import { existsSync, lstatSync, readdirSync, readFileSync, rmSync, unlinkSync } from "node:fs";import { basename, dirname, join, resolve } from "node:path";import chalk from "chalk";import { APP_NAME, getAgentDir, VERSION } from "../config.js";2 unmodified lines killOrphanProcess, readActiveOrphanProcesses, shouldReapOrphanProcess,} from "../core/orphan-process-journal.js";import { getProcessStartId } from "../core/session-lease.js";import { canonicalSessionPath, getProcessStartId } from "../core/session-lease.js";import { DaemonClient } from "../modes/daemon/daemon-client.js";import { DAEMON_PROTOCOL_VERSION, DAEMON_SCHEMA_ID,1000 unmodified linesExpand all normalizeSocketPath(worker.descriptor.supervisorSocketPath) === normalizeSocketPath(supervisorSocketPath), );}
function findAllTrackedWorkers(): TrackedWorker[] { const root = join(getAgentDir(), "daemon-workers");function findAllTrackedWorkers(agentDir: string = getAgentDir()): TrackedWorker[] { const root = join(agentDir, "daemon-workers"); if (!existsSync(root)) { return []; } const workers: TrackedWorker[] = [];100 unmodified lines } return getProcessStartId(pid) === expectedStartId;}
export function reapStaleWorkerSessions( agentDir: string = getAgentDir(),): Array<{ socketPath: string; action: string }> { const reaped: Array<{ socketPath: string; action: string }> = []; const leaseRoot = join(agentDir, "session-leases");
for (const { descriptor, descriptorPath } of findAllTrackedWorkers(agentDir)) { const isDead = !isProcessAlive(descriptor.pid) || (descriptor.processStartId !== undefined && !trackedLeaderIdentityCurrent(descriptor.pid, descriptor.processStartId)); if (!isDead || (descriptor.lifecycle !== "recovering" && (descriptor.consecutiveFailures ?? 0) === 0)) continue; const sessionFile = descriptor.sessionFile ?? descriptor.createCommand?.sessionPath; const sessionName = sessionFile ? basename(sessionFile).replace(/\.jsonl$/, "") : (descriptor.rootSessionId ?? descriptor.rootActiveSessionId ?? "unknown");
rmSync(descriptorPath, { force: true }); if (descriptor.recoveryJournalPath) rmSync(descriptor.recoveryJournalPath, { force: true }); if (descriptor.orphanProcessJournalPath) rmSync(descriptor.orphanProcessJournalPath, { force: true });
if (sessionFile && existsSync(leaseRoot)) { const leaseKey = createHash("sha256").update(canonicalSessionPath(sessionFile)).digest("hex"); const leaseDir = join(leaseRoot, `${leaseKey}.lock`); rmSync(leaseDir, { recursive: true, force: true }); rmSync(`${leaseDir}.guard`, { force: true }); } if (descriptor.socketPath) rmSync(descriptor.socketPath, { force: true });
reaped.push({ socketPath: `session ${sessionName}`, action: "cleared stale recovering worker, lease, and socket", }); }
if (existsSync(leaseRoot)) { let entries: string[] = []; try { entries = readdirSync(leaseRoot); } catch {} for (const entry of entries) { if (!entry.endsWith(".lock")) continue; const dir = join(leaseRoot, entry); const ownerPath = join(dir, "owner.json"); if (!existsSync(ownerPath)) continue; try { const owner = JSON.parse(readFileSync(ownerPath, "utf8")) as { pid?: number; processStartId?: string; sessionPath?: string; }; if ( typeof owner.pid === "number" && (!isProcessAlive(owner.pid) || (owner.processStartId !== undefined && !trackedLeaderIdentityCurrent(owner.pid, owner.processStartId))) ) { rmSync(dir, { recursive: true, force: true }); rmSync(`${dir}.guard`, { force: true }); const sName = owner.sessionPath ? basename(owner.sessionPath).replace(/\.jsonl$/, "") : entry; if (!reaped.some((r) => r.socketPath === `session ${sName}`)) { reaped.push({ socketPath: `session ${sName}`, action: "cleared stale session lease" }); } } } catch {} } }
return reaped;}
export async function runReap(json: boolean, force: boolean): Promise<void> { const daemons = await discoverDaemons(); const reaped: Array<{ socketPath: string; action: string }> = []; const reaped: Array<{ socketPath: string; action: string }> = [...reapStaleWorkerSessions()]; const skipped: Array<{ socketPath: string; reason: string }> = [];
for (const action of planReap(daemons, force)) { const { socketPath, pid } = action.daemon;168 unmodified linesExpand all