From cb9a6a1cd37b4e84c34e778dc9be79c6238cb83e Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 7 Aug 2026 15:09:19 +0200 Subject: [PATCH 01/13] fix(coding-agent): finalize timed-out worker stops instead of stranding registrations When a worker does not exit within the stop deadline, the supervisor now keeps watching the process, escalates to SIGKILL, and completes the interrupted cleanup once the process dies. Process liveness checks also treat zombie processes as dead so cleanup is not deferred forever. --- .../src/modes/daemon/daemon-supervisor.ts | 60 ++++++-- .../coding-agent/src/utils/child-process.ts | 36 ++++- .../coding-agent/test/child-process.test.ts | 50 ++++++- .../test/daemon-supervisor-monitor.test.ts | 135 ++++++++++++++++++ .../test/daemon-supervisor-process.test.ts | 65 +++++++++ 5 files changed, 333 insertions(+), 13 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index c7bb17d1c..aba3113fe 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -43,7 +43,7 @@ import { import { canonicalSessionPath, getProcessStartId, SessionAlreadyActiveError } from "../../core/session-lease.js"; import { readSessionInfo, type SessionInfo } from "../../core/session-manager.js"; import { SettingsManager } from "../../core/settings-manager.js"; -import { signalProcessGroupOrProcess } from "../../utils/child-process.js"; +import { isProcessAlive, signalProcessGroupOrProcess } from "../../utils/child-process.js"; import type { AgentConnectionHeartbeat } from "../agent-connection/types.js"; import { attachJsonlLineReader, serializeJsonLine } from "../rpc/jsonl.js"; import type { PrivateFrame } from "../session-worker/private-framing.js"; @@ -136,6 +136,8 @@ const UPDATE_RESTART_WORKER_REQUEST_TIMEOUT_MS = 90_000; const UPDATE_RESTART_PREPARE_DEADLINE_MS = 100_000; const WORKER_RETRY_DELAYS_MS = [250, 1000, 5000] as const; const DEFERRED_RECOVERY_RECHECK_MS = 5000; +const STOP_FINALIZATION_RECHECK_MS = 250; +const STOP_FINALIZATION_SIGKILL_GRACE_MS = 5000; const OWNED_WORKER_DISCONNECT_GRACE_MS = 30_000; const IDLE_EVICTION_MAX_SWEEP_INTERVAL_MS = 5 * 60_000; const IDLE_EVICTION_MIN_SWEEP_INTERVAL_MS = 60_000; @@ -261,6 +263,7 @@ interface ResidentWorker { intentionalStop: boolean; stopRevision: number; launchEnv?: Record; + stopFinalization?: Promise; ownerCleanupTimer?: ReturnType; promotedOwnerClientId?: string; updateRestartPrepareClient?: DaemonWorkerClient; @@ -511,15 +514,6 @@ function looksLikeSessionPath(selector: string): boolean { return isAbsolute(selector) || selector.endsWith(".jsonl") || selector.includes("/") || selector.includes("\\"); } -function isProcessAlive(pid: number): boolean { - try { - process.kill(pid, 0); - return true; - } catch (error) { - return (error as NodeJS.ErrnoException).code === "EPERM"; - } -} - function isFinalizedTranscriptEvent(eventType: string | undefined): boolean { return ( eventType === "message_end" || @@ -4648,6 +4642,9 @@ export class DaemonSupervisor { } if (isWorkerProcessAlive()) { worker.intentionalStop = worker.descriptor.stopRequestedAt !== undefined; + if (removeDescriptor) { + this.scheduleWorkerStopFinalization(worker); + } throw new Error(`Session worker ${worker.descriptor.workerId} did not stop${force ? " after SIGKILL" : ""}`); } if (directChild) { @@ -4669,6 +4666,49 @@ export class DaemonSupervisor { } } + /** + * A stop that timed out leaves a tombstoned registration behind. Keep + * escalating in the background until the process is gone, then finish the + * interrupted cleanup instead of leaving a dead worker registered forever. + */ + private scheduleWorkerStopFinalization(worker: ResidentWorker): void { + if (worker.stopFinalization) { + return; + } + worker.stopFinalization = this.finalizeTimedOutWorkerStop(worker).finally(() => { + worker.stopFinalization = undefined; + }); + } + + private async finalizeTimedOutWorkerStop(worker: ResidentWorker): Promise { + const sigkillDeadline = Date.now() + STOP_FINALIZATION_SIGKILL_GRACE_MS; + let killed = false; + while (!this.shuttingDown) { + if (!isProcessAlive(worker.descriptor.pid)) { + break; + } + if (!killed && Date.now() >= sigkillDeadline) { + signalProcessGroupOrProcess(worker.descriptor.pid, "SIGKILL"); + killed = true; + } + await unrefDelay(STOP_FINALIZATION_RECHECK_MS); + } + if (this.shuttingDown || this.workers.get(worker.descriptor.workerId) !== worker) { + return; + } + if (worker.descriptor.stopRequestedAt === undefined) { + // The stop was rescinded (for example by an explicit retry) while the + // process was still exiting; leave the registration to that flow. + return; + } + try { + await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); + this.log(`Finalized timed-out stop for worker ${worker.descriptor.workerId}`); + } catch (error) { + this.reportCleanupFailure(`timed-out worker stop ${worker.descriptor.workerId}`, error); + } + } + private async finalizeArchivedWorkerStop(worker: ResidentWorker): Promise { const context = this.workerSessionArtifactContext(worker); if (!context) { diff --git a/packages/coding-agent/src/utils/child-process.ts b/packages/coding-agent/src/utils/child-process.ts index dd2d02134..4ef6d55e3 100644 --- a/packages/coding-agent/src/utils/child-process.ts +++ b/packages/coding-agent/src/utils/child-process.ts @@ -1,4 +1,5 @@ -import type { ChildProcess } from "node:child_process"; +import { type ChildProcess, execFileSync } from "node:child_process"; +import { readFileSync } from "node:fs"; import { constants } from "node:os"; import { basename } from "node:path"; @@ -12,6 +13,39 @@ export function shouldUseWindowsShell(command: string): boolean { return commandName.endsWith(".cmd") || commandName.endsWith(".bat") || WINDOWS_SHELL_COMMANDS.has(commandName); } +/** A zombie has already exited; it only lingers until its parent reaps it. */ +export function isZombieProcess(pid: number): boolean { + if (process.platform === "win32") { + return false; + } + try { + const stat = readFileSync(`/proc/${pid}/stat`, "utf8"); + const state = stat + .slice(stat.lastIndexOf(")") + 2) + .trimStart() + .charAt(0); + return state === "Z"; + } catch { + // Fall through to the portable process listing used on macOS and BSD. + } + try { + const state = execFileSync("ps", ["-p", String(pid), "-o", "stat="], { encoding: "utf8" }).trim(); + return state.startsWith("Z"); + } catch { + return false; + } +} + +/** True only for a process that is actually running: zombies do not count. */ +export function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } + return !isZombieProcess(pid); +} + export function signalProcessGroupOrProcess(pid: number, signal: NodeJS.Signals): void { try { process.kill(-pid, signal); diff --git a/packages/coding-agent/test/child-process.test.ts b/packages/coding-agent/test/child-process.test.ts index 8000c2f9c..6a807f32f 100644 --- a/packages/coding-agent/test/child-process.test.ts +++ b/packages/coding-agent/test/child-process.test.ts @@ -1,7 +1,7 @@ -import type { ChildProcess } from "node:child_process"; +import { type ChildProcess, spawn } from "node:child_process"; import { EventEmitter } from "node:events"; import { describe, expect, it } from "vitest"; -import { waitForChildProcess } from "../src/utils/child-process.js"; +import { isProcessAlive, isZombieProcess, waitForChildProcess } from "../src/utils/child-process.js"; describe("waitForChildProcess", () => { it("reports signaled already-exited children as failures", async () => { @@ -15,3 +15,49 @@ describe("waitForChildProcess", () => { await expect(waitForChildProcess(child as unknown as ChildProcess)).resolves.toBe(143); }); }); + +describe("process liveness", () => { + it("treats the current process as alive and not a zombie", () => { + expect(isProcessAlive(process.pid)).toBe(true); + expect(isZombieProcess(process.pid)).toBe(false); + }); + + it("treats an exited process as dead", async () => { + const child = spawn(process.execPath, ["--eval", "process.exit(0)"], { stdio: "ignore" }); + await new Promise((resolveExit) => child.once("exit", () => resolveExit())); + // Node reaps its own children on exit, so the pid is fully gone. + await new Promise((resolveDelay) => setTimeout(resolveDelay, 50)); + expect(isProcessAlive(child.pid!)).toBe(false); + }); + + it.skipIf(process.platform === "win32")("treats a zombie process as dead", async () => { + // A parent that forks and never reaps leaves the child as a zombie. + const parent = spawn( + "perl", + ["-e", '$| = 1; my $pid = fork(); if ($pid) { print "$pid\\n"; sleep 30 } else { exit 0 }'], + { stdio: ["ignore", "pipe", "ignore"] }, + ); + try { + const zombiePid = await new Promise((resolvePid, rejectPid) => { + let output = ""; + const timer = setTimeout(() => rejectPid(new Error("Timed out waiting for the zombie pid")), 5000); + parent.stdout.on("data", (chunk: Buffer) => { + output += chunk.toString(); + const parsed = Number.parseInt(output.trim(), 10); + if (Number.isInteger(parsed) && parsed > 0) { + clearTimeout(timer); + resolvePid(parsed); + } + }); + }); + const deadline = Date.now() + 5000; + while (!isZombieProcess(zombiePid) && Date.now() < deadline) { + await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); + } + expect(isZombieProcess(zombiePid)).toBe(true); + expect(isProcessAlive(zombiePid)).toBe(false); + } finally { + parent.kill("SIGKILL"); + } + }); +}); diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 112d75487..843bcd3df 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1699,6 +1699,141 @@ describe("daemon worker supervisor monitoring", () => { ]); }); + it("finalizes a timed-out stop once the worker process dies", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-timed-out-stop", + pid: process.pid, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + let alive = true; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + stopWorker, + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockImplementation(() => alive); + try { + supervisor.scheduleWorkerStopFinalization(worker); + const finalization = worker.stopFinalization; + expect(finalization).toBeDefined(); + + await vi.advanceTimersByTimeAsync(500); + expect(stopWorker).not.toHaveBeenCalled(); + + alive = false; + await vi.advanceTimersByTimeAsync(500); + await finalization; + + expect(stopWorker).toHaveBeenCalledWith(worker, true, true, false); + expect(worker.stopFinalization).toBeUndefined(); + } finally { + aliveSpy.mockRestore(); + } + }); + + it("escalates a stuck stop to SIGKILL before finalizing", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-stuck-stop", + pid: process.pid, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + archiveOnStop: true, + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + let alive = true; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + stopWorker, + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockImplementation(() => alive); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation((_pid, signal) => { + if (signal === "SIGKILL") { + alive = false; + } + }); + try { + supervisor.scheduleWorkerStopFinalization(worker); + const finalization = worker.stopFinalization; + + await vi.advanceTimersByTimeAsync(10_000); + await finalization; + + expect(killSpy).toHaveBeenCalledWith(worker.descriptor.pid, "SIGKILL"); + expect(stopWorker).toHaveBeenCalledWith(worker, true, true, true); + } finally { + aliveSpy.mockRestore(); + killSpy.mockRestore(); + } + }); + + it("leaves a rescinded stop to the retry flow instead of finalizing it", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-rescinded-stop", + pid: process.pid, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString() as string | undefined, + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + let alive = true; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + stopWorker, + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockImplementation(() => alive); + try { + supervisor.scheduleWorkerStopFinalization(worker); + const finalization = worker.stopFinalization; + + // An explicit retry revives the worker while its process is exiting. + worker.descriptor.stopRequestedAt = undefined; + worker.intentionalStop = false; + alive = false; + await vi.advanceTimersByTimeAsync(500); + await finalization; + + expect(stopWorker).not.toHaveBeenCalled(); + } finally { + aliveSpy.mockRestore(); + } + }); + it("ignores malformed persisted worker descriptors", () => { const descriptorDir = mkdtempSync(join(tmpdir(), "prime-supervisor-descriptor-test-")); try { diff --git a/packages/coding-agent/test/daemon-supervisor-process.test.ts b/packages/coding-agent/test/daemon-supervisor-process.test.ts index e013e9d13..3f2d7d8fd 100644 --- a/packages/coding-agent/test/daemon-supervisor-process.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-process.test.ts @@ -827,6 +827,71 @@ describe("daemon supervisor resident workers", () => { await waitForSocketGone(socketPath); }, 30_000); + it( + "finalizes a timed-out worker stop by force-stopping the process and removing its registration", + { tags: ["process-stress"], timeout: 45_000 }, + async () => { + const root = tempDir(); + const agentDir = join(root, "agent"); + const projectDir = join(root, "project"); + const sessionDir = join(agentDir, "sessions"); + const socketPath = join( + tmpdir(), + `prime-supervisor-stop-finalize-${process.pid}-${randomUUID().slice(0, 8)}.sock`, + ); + mkdirSync(projectDir, { recursive: true }); + const sessionManager = SessionManager.create(projectDir, sessionDir); + sessionManager.appendMessage({ role: "user", content: "finalize me", timestamp: 1 }); + const sessionFile = sessionManager.getSessionFile(); + if (!sessionFile) { + throw new Error("Fixture session did not persist"); + } + + const supervisor = spawnSupervisor(agentDir, socketPath, projectDir); + const client = await connectEventually(socketPath, supervisor); + const created = await client.request({ + type: "create", + sessionPath: sessionFile, + lifecycle: "client_owned", + config: { cwd: projectDir, agentDir, sessionDir, noTools: true, noExtensions: true }, + }); + if (!created.success) { + throw new Error(created.error); + } + const summary = requireSummary(created.data); + if (!summary.workerPid) { + throw new Error("Resident worker did not expose its pid"); + } + workerPids.add(summary.workerPid); + const activeSessionId = summary.activeSessionId ?? summary.id; + + // A suspended worker cannot exit within the stop deadline, so the stop + // times out and used to leave a tombstoned registration behind forever. + process.kill(summary.workerPid, "SIGSTOP"); + const stopResult = await client.request({ type: "complete_owned_session", activeSessionId }, 30_000); + expect(stopResult).toMatchObject({ + success: false, + error: expect.stringContaining("did not stop"), + }); + const tombstone = readWorkerDescriptor(agentDir); + expect(tombstone.stopRequestedAt).toEqual(expect.any(String)); + + // The supervisor finishes the interrupted stop on its own: it escalates + // to SIGKILL, waits for the process to die, and removes the registration. + await waitForProcessGone(summary.workerPid); + workerPids.delete(summary.workerPid); + await waitForCondition( + () => countWorkerDescriptors(agentDir) === 0, + "Timed-out worker stop was not finalized", + 20_000, + ); + + await client.request({ type: "shutdown" }); + client.close(); + await waitForSocketGone(socketPath); + }, + ); + it( "does not resurrect an intentionally stopped root when the supervisor dies during kill", { tags: ["process-stress"], timeout: 30_000 }, From 0fd099112ac0ed33c5d5d51892299cf39c593303 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 7 Aug 2026 15:52:01 +0200 Subject: [PATCH 02/13] fix(coding-agent): bind stop finalization to the exact worker process generation The background finalizer now snapshots pid, processStartId, and stopRevision when scheduled and aborts if the stop is rescinded or the worker is relaunched, so it can never SIGKILL a retried worker or an unrelated process that reused the pid. stopWorker signalling is likewise identity-aware. --- .../src/modes/daemon/daemon-supervisor.ts | 52 ++++++++-- .../test/daemon-supervisor-monitor.test.ts | 99 +++++++++++++++++++ 2 files changed, 141 insertions(+), 10 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index aba3113fe..000954321 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4535,6 +4535,18 @@ export class DaemonSupervisor { renameSync(tempPath, path); } + /** True when the registered pid is alive and still the process we launched. */ + private isWorkerProcessCurrent(worker: ResidentWorker): boolean { + if (!isProcessAlive(worker.descriptor.pid)) { + return false; + } + if (worker.descriptor.processStartId === undefined) { + return true; + } + const observed = getProcessStartId(worker.descriptor.pid); + return observed === undefined || observed === worker.descriptor.processStartId; + } + private async stopWorker( worker: ResidentWorker, removeDescriptor: boolean, @@ -4618,13 +4630,14 @@ export class DaemonSupervisor { worker.client = undefined; } else if (directChild) { directChild.child.kill("SIGTERM"); - } else if (isProcessAlive(worker.descriptor.pid)) { + } else if (this.isWorkerProcessCurrent(worker)) { signalProcessGroupOrProcess(worker.descriptor.pid, "SIGTERM"); } + // Identity-aware: a recycled pid is treated as gone and never signalled. const isWorkerProcessAlive = () => directChild ? directChild.child.exitCode === null && directChild.child.signalCode === null - : isProcessAlive(worker.descriptor.pid); + : this.isWorkerProcessCurrent(worker); const gracefulDeadline = Date.now() + (force ? 500 : 2000); while (isWorkerProcessAlive() && Date.now() < gracefulDeadline) { await delay(25); @@ -4681,24 +4694,43 @@ export class DaemonSupervisor { } private async finalizeTimedOutWorkerStop(worker: ResidentWorker): Promise { + // Bind to the exact process generation being stopped: a retry can rescind + // the stop and relaunch with a new pid, and the OS can recycle the old + // pid. The finalizer must never follow either successor. + const pid = worker.descriptor.pid; + const processStartId = worker.descriptor.processStartId ?? getProcessStartId(pid); + const stopRevision = worker.stopRevision; + const isStopGenerationCurrent = () => + this.workers.get(worker.descriptor.workerId) === worker && + worker.stopRevision === stopRevision && + worker.descriptor.stopRequestedAt !== undefined && + worker.descriptor.pid === pid; + const isStoppedProcessAlive = () => { + if (!isProcessAlive(pid)) { + return false; + } + if (processStartId === undefined) { + return true; + } + const observed = getProcessStartId(pid); + return observed === undefined || observed === processStartId; + }; const sigkillDeadline = Date.now() + STOP_FINALIZATION_SIGKILL_GRACE_MS; let killed = false; while (!this.shuttingDown) { - if (!isProcessAlive(worker.descriptor.pid)) { + if (!isStopGenerationCurrent()) { + return; + } + if (!isStoppedProcessAlive()) { break; } if (!killed && Date.now() >= sigkillDeadline) { - signalProcessGroupOrProcess(worker.descriptor.pid, "SIGKILL"); + signalProcessGroupOrProcess(pid, "SIGKILL"); killed = true; } await unrefDelay(STOP_FINALIZATION_RECHECK_MS); } - if (this.shuttingDown || this.workers.get(worker.descriptor.workerId) !== worker) { - return; - } - if (worker.descriptor.stopRequestedAt === undefined) { - // The stop was rescinded (for example by an explicit retry) while the - // process was still exiting; leave the registration to that flow. + if (this.shuttingDown || !isStopGenerationCurrent()) { return; } try { diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 843bcd3df..2a6747159 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1791,6 +1791,105 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("never follows a relaunched worker pid after a retry rescinds the stop", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-relaunched", + pid: 111_111, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString() as string | undefined, + }, + intentionalStop: true, + stopRevision: 3, + stopFinalization: undefined as Promise | undefined, + }; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + stopWorker, + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + try { + supervisor.scheduleWorkerStopFinalization(worker); + const finalization = worker.stopFinalization; + + // An explicit retry rescinds the stop and relaunches with a new pid + // while the old process is still wedged. + await vi.advanceTimersByTimeAsync(1000); + worker.descriptor.stopRequestedAt = undefined; + worker.intentionalStop = false; + worker.stopRevision = 4; + worker.descriptor.pid = 222_222; + + await vi.advanceTimersByTimeAsync(20_000); + await finalization; + + // The healthy relaunched worker must never be signalled or stopped. + expect(killSpy).not.toHaveBeenCalledWith(222_222, "SIGKILL"); + expect(killSpy).not.toHaveBeenCalled(); + expect(stopWorker).not.toHaveBeenCalled(); + } finally { + aliveSpy.mockRestore(); + killSpy.mockRestore(); + } + }); + + it("treats a recycled pid as gone instead of killing its new owner", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-recycled-pid", + pid: 111_112, + processStartId: "proc:original", + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + stopWorker, + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const sessionLeaseModule = await import("../src/core/session-lease.js"); + // The pid is alive, but it now belongs to an unrelated process. + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue("proc:recycled"); + try { + supervisor.scheduleWorkerStopFinalization(worker); + const finalization = worker.stopFinalization; + + await vi.advanceTimersByTimeAsync(20_000); + await finalization; + + // The original worker is gone, so the stop is finalized without ever + // signalling the unrelated pid owner. + expect(killSpy).not.toHaveBeenCalled(); + expect(stopWorker).toHaveBeenCalledWith(worker, true, true, false); + } finally { + aliveSpy.mockRestore(); + killSpy.mockRestore(); + startIdSpy.mockRestore(); + } + }); + it("leaves a rescinded stop to the retry flow instead of finalizing it", async () => { vi.useFakeTimers(); const worker = { From d0cd7bc9e7e2f50cf5715efb99f075cbe76efe63 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 7 Aug 2026 17:02:17 +0200 Subject: [PATCH 03/13] fix(coding-agent): harden stop finalization identity checks and retries - Fail closed when a recorded processStartId cannot be observed, so a recycled pid is never signalled even if identity observation fails. - Record a schedule-time identity for workers that never had one. - Retry transient finalization cleanup failures instead of stranding the dead registration permanently. - Probe liveness with a cheap kill(0) on every poll and throttle the ps-backed zombie/identity checks so wedged workers cannot saturate the supervisor event loop. --- .../src/modes/daemon/daemon-supervisor.ts | 101 ++++++++++++++---- .../coding-agent/src/utils/child-process.ts | 17 +-- .../test/daemon-supervisor-monitor.test.ts | 58 ++++++++++ 3 files changed, 148 insertions(+), 28 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 000954321..7730132b5 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -43,7 +43,7 @@ import { import { canonicalSessionPath, getProcessStartId, SessionAlreadyActiveError } from "../../core/session-lease.js"; import { readSessionInfo, type SessionInfo } from "../../core/session-manager.js"; import { SettingsManager } from "../../core/settings-manager.js"; -import { isProcessAlive, signalProcessGroupOrProcess } from "../../utils/child-process.js"; +import { isProcessAlive, processIdExists, signalProcessGroupOrProcess } from "../../utils/child-process.js"; import type { AgentConnectionHeartbeat } from "../agent-connection/types.js"; import { attachJsonlLineReader, serializeJsonLine } from "../rpc/jsonl.js"; import type { PrivateFrame } from "../session-worker/private-framing.js"; @@ -138,6 +138,11 @@ const WORKER_RETRY_DELAYS_MS = [250, 1000, 5000] as const; const DEFERRED_RECOVERY_RECHECK_MS = 5000; const STOP_FINALIZATION_RECHECK_MS = 250; const STOP_FINALIZATION_SIGKILL_GRACE_MS = 5000; +const STOP_FINALIZATION_RETRY_MS = 5000; +// Polling loops probe existence cheaply via kill(0); the ps-backed zombie and +// identity checks are throttled so a wedged worker cannot saturate the +// supervisor event loop with synchronous subprocess spawns. +const LIVENESS_IDENTITY_RECHECK_MS = 500; const OWNED_WORKER_DISCONNECT_GRACE_MS = 30_000; const IDLE_EVICTION_MAX_SWEEP_INTERVAL_MS = 5 * 60_000; const IDLE_EVICTION_MIN_SWEEP_INTERVAL_MS = 60_000; @@ -4535,7 +4540,11 @@ export class DaemonSupervisor { renameSync(tempPath, path); } - /** True when the registered pid is alive and still the process we launched. */ + /** + * True when the registered pid is alive and still the process we launched. + * With a recorded identity this fails closed: an unavailable observation is + * treated as a different process so a recycled pid is never signalled. + */ private isWorkerProcessCurrent(worker: ResidentWorker): boolean { if (!isProcessAlive(worker.descriptor.pid)) { return false; @@ -4543,8 +4552,7 @@ export class DaemonSupervisor { if (worker.descriptor.processStartId === undefined) { return true; } - const observed = getProcessStartId(worker.descriptor.pid); - return observed === undefined || observed === worker.descriptor.processStartId; + return getProcessStartId(worker.descriptor.pid) === worker.descriptor.processStartId; } private async stopWorker( @@ -4634,10 +4642,23 @@ export class DaemonSupervisor { signalProcessGroupOrProcess(worker.descriptor.pid, "SIGTERM"); } // Identity-aware: a recycled pid is treated as gone and never signalled. - const isWorkerProcessAlive = () => - directChild - ? directChild.child.exitCode === null && directChild.child.signalCode === null - : this.isWorkerProcessCurrent(worker); + // kill(0) runs on every poll; the expensive identity check is throttled. + let identityVerdict = true; + let identityCheckedAt = 0; + const isWorkerProcessAlive = () => { + if (directChild) { + return directChild.child.exitCode === null && directChild.child.signalCode === null; + } + if (!processIdExists(worker.descriptor.pid)) { + return false; + } + const now = Date.now(); + if (now - identityCheckedAt >= LIVENESS_IDENTITY_RECHECK_MS) { + identityCheckedAt = now; + identityVerdict = this.isWorkerProcessCurrent(worker); + } + return identityVerdict; + }; const gracefulDeadline = Date.now() + (force ? 500 : 2000); while (isWorkerProcessAlive() && Date.now() < gracefulDeadline) { await delay(25); @@ -4698,22 +4719,49 @@ export class DaemonSupervisor { // the stop and relaunch with a new pid, and the OS can recycle the old // pid. The finalizer must never follow either successor. const pid = worker.descriptor.pid; - const processStartId = worker.descriptor.processStartId ?? getProcessStartId(pid); + let processStartId = worker.descriptor.processStartId; + if (processStartId === undefined) { + // The stop timed out because the process was still alive moments ago, + // so an identity observed now can be trusted and recorded. It lets the + // eventual stopWorker call fail closed on a recycled pid too. + const observed = getProcessStartId(pid); + if (observed !== undefined && isProcessAlive(pid)) { + processStartId = observed; + worker.descriptor.processStartId = observed; + try { + this.persistWorker(worker); + } catch (error) { + this.reportCleanupFailure(`worker identity record ${worker.descriptor.workerId}`, error); + } + } + } const stopRevision = worker.stopRevision; const isStopGenerationCurrent = () => this.workers.get(worker.descriptor.workerId) === worker && worker.stopRevision === stopRevision && worker.descriptor.stopRequestedAt !== undefined && worker.descriptor.pid === pid; + let stoppedVerdict = true; + let stoppedCheckedAt = 0; const isStoppedProcessAlive = () => { - if (!isProcessAlive(pid)) { + if (!processIdExists(pid)) { return false; } - if (processStartId === undefined) { - return true; + const now = Date.now(); + if (now - stoppedCheckedAt < LIVENESS_IDENTITY_RECHECK_MS) { + return stoppedVerdict; } - const observed = getProcessStartId(pid); - return observed === undefined || observed === processStartId; + stoppedCheckedAt = now; + if (!isProcessAlive(pid)) { + stoppedVerdict = false; + } else if (processStartId === undefined) { + stoppedVerdict = true; + } else { + // Fail closed: an unobservable identity is treated as a different + // process so a recycled pid is never SIGKILLed. + stoppedVerdict = getProcessStartId(pid) === processStartId; + } + return stoppedVerdict; }; const sigkillDeadline = Date.now() + STOP_FINALIZATION_SIGKILL_GRACE_MS; let killed = false; @@ -4730,14 +4778,23 @@ export class DaemonSupervisor { } await unrefDelay(STOP_FINALIZATION_RECHECK_MS); } - if (this.shuttingDown || !isStopGenerationCurrent()) { - return; - } - try { - await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); - this.log(`Finalized timed-out stop for worker ${worker.descriptor.workerId}`); - } catch (error) { - this.reportCleanupFailure(`timed-out worker stop ${worker.descriptor.workerId}`, error); + // Retry transient cleanup failures (for example catalog archival) so a + // dead worker's registration is never stranded permanently. Each attempt + // bumps the worker's stopRevision, so rescission is detected through the + // registration and tombstone instead of the waiting-phase snapshot. + const isCleanupStillWanted = () => + this.workers.get(worker.descriptor.workerId) === worker && + worker.descriptor.stopRequestedAt !== undefined && + worker.descriptor.pid === pid; + while (!this.shuttingDown && isCleanupStillWanted()) { + try { + await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); + this.log(`Finalized timed-out stop for worker ${worker.descriptor.workerId}`); + return; + } catch (error) { + this.reportCleanupFailure(`timed-out worker stop ${worker.descriptor.workerId}`, error); + await unrefDelay(STOP_FINALIZATION_RETRY_MS); + } } } diff --git a/packages/coding-agent/src/utils/child-process.ts b/packages/coding-agent/src/utils/child-process.ts index 4ef6d55e3..3f587327f 100644 --- a/packages/coding-agent/src/utils/child-process.ts +++ b/packages/coding-agent/src/utils/child-process.ts @@ -13,6 +13,16 @@ export function shouldUseWindowsShell(command: string): boolean { return commandName.endsWith(".cmd") || commandName.endsWith(".bat") || WINDOWS_SHELL_COMMANDS.has(commandName); } +/** Cheap kill(0) existence probe; counts zombies as existing. */ +export function processIdExists(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + /** A zombie has already exited; it only lingers until its parent reaps it. */ export function isZombieProcess(pid: number): boolean { if (process.platform === "win32") { @@ -38,12 +48,7 @@ export function isZombieProcess(pid: number): boolean { /** True only for a process that is actually running: zombies do not count. */ export function isProcessAlive(pid: number): boolean { - try { - process.kill(pid, 0); - } catch (error) { - return (error as NodeJS.ErrnoException).code === "EPERM"; - } - return !isZombieProcess(pid); + return processIdExists(pid) && !isZombieProcess(pid); } export function signalProcessGroupOrProcess(pid: number, signal: NodeJS.Signals): void { diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 2a6747159..4565d745c 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1718,6 +1718,7 @@ describe("daemon worker supervisor monitoring", () => { workers: new Map([[worker.descriptor.workerId, worker]]), shuttingDown: false, stopWorker, + persistWorker: vi.fn(), log: vi.fn(), reportCleanupFailure: vi.fn(), }) as { @@ -1764,6 +1765,7 @@ describe("daemon worker supervisor monitoring", () => { workers: new Map([[worker.descriptor.workerId, worker]]), shuttingDown: false, stopWorker, + persistWorker: vi.fn(), log: vi.fn(), reportCleanupFailure: vi.fn(), }) as { @@ -1809,12 +1811,14 @@ describe("daemon worker supervisor monitoring", () => { workers: new Map([[worker.descriptor.workerId, worker]]), shuttingDown: false, stopWorker, + persistWorker: vi.fn(), log: vi.fn(), reportCleanupFailure: vi.fn(), }) as { scheduleWorkerStopFinalization(target: object): void; }; const childProcessModule = await import("../src/utils/child-process.js"); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); try { @@ -1837,6 +1841,7 @@ describe("daemon worker supervisor monitoring", () => { expect(killSpy).not.toHaveBeenCalled(); expect(stopWorker).not.toHaveBeenCalled(); } finally { + existsSpy.mockRestore(); aliveSpy.mockRestore(); killSpy.mockRestore(); } @@ -1861,6 +1866,7 @@ describe("daemon worker supervisor monitoring", () => { workers: new Map([[worker.descriptor.workerId, worker]]), shuttingDown: false, stopWorker, + persistWorker: vi.fn(), log: vi.fn(), reportCleanupFailure: vi.fn(), }) as { @@ -1869,6 +1875,7 @@ describe("daemon worker supervisor monitoring", () => { const childProcessModule = await import("../src/utils/child-process.js"); const sessionLeaseModule = await import("../src/core/session-lease.js"); // The pid is alive, but it now belongs to an unrelated process. + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue("proc:recycled"); @@ -1884,12 +1891,62 @@ describe("daemon worker supervisor monitoring", () => { expect(killSpy).not.toHaveBeenCalled(); expect(stopWorker).toHaveBeenCalledWith(worker, true, true, false); } finally { + existsSpy.mockRestore(); aliveSpy.mockRestore(); killSpy.mockRestore(); startIdSpy.mockRestore(); } }); + it("retries finalization after a transient cleanup failure", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-transient-cleanup", + pid: process.pid, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + const workers = new Map([[worker.descriptor.workerId, worker]]); + const stopWorker = vi + .fn(async () => { + workers.delete(worker.descriptor.workerId); + }) + .mockImplementationOnce(async () => { + throw new Error("archive temporarily unavailable"); + }); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers, + shuttingDown: false, + stopWorker, + persistWorker: vi.fn(), + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(false); + try { + supervisor.scheduleWorkerStopFinalization(worker); + const finalization = worker.stopFinalization; + + await vi.advanceTimersByTimeAsync(20_000); + await finalization; + + // The first attempt failed transiently; the registration is still + // cleaned up by a retry instead of being stranded forever. + expect(stopWorker).toHaveBeenCalledTimes(2); + expect(workers.size).toBe(0); + } finally { + aliveSpy.mockRestore(); + } + }); + it("leaves a rescinded stop to the retry flow instead of finalizing it", async () => { vi.useFakeTimers(); const worker = { @@ -1909,6 +1966,7 @@ describe("daemon worker supervisor monitoring", () => { workers: new Map([[worker.descriptor.workerId, worker]]), shuttingDown: false, stopWorker, + persistWorker: vi.fn(), log: vi.fn(), reportCleanupFailure: vi.fn(), }) as { From 92c6853a14fd3eeb36548dfe225eedf326df7c6e Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 7 Aug 2026 18:19:56 +0200 Subject: [PATCH 04/13] fix(coding-agent): treat unobservable process identity as alive, not gone stopWorker used the identity check as a liveness predicate, so a transient getProcessStartId failure could skip signalling and delete the registration of a still-running worker. Identity verdicts are now directional: only a confirmed-current pid is signalled, only a confirmed-gone/replaced pid is cleaned up, and an unknown verdict keeps waiting. --- .../src/modes/daemon/daemon-supervisor.ts | 49 ++++++++++++------- .../test/daemon-supervisor-monitor.test.ts | 49 +++++++++++++++++++ 2 files changed, 80 insertions(+), 18 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 7730132b5..4b092ccd3 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4541,18 +4541,24 @@ export class DaemonSupervisor { } /** - * True when the registered pid is alive and still the process we launched. - * With a recorded identity this fails closed: an unavailable observation is - * treated as a different process so a recycled pid is never signalled. + * Verdict on whether the registered pid is still the process we launched. + * Callers must be conservative in both directions: signal a pid only on + * "current" (never SIGKILL a recycled pid), and clean up a registration + * only on "gone"/"replaced" (never orphan a live worker because a + * transient identity lookup failed). */ - private isWorkerProcessCurrent(worker: ResidentWorker): boolean { + private workerProcessIdentity(worker: ResidentWorker): "current" | "replaced" | "gone" | "unknown" { if (!isProcessAlive(worker.descriptor.pid)) { - return false; + return "gone"; } if (worker.descriptor.processStartId === undefined) { - return true; + return "current"; } - return getProcessStartId(worker.descriptor.pid) === worker.descriptor.processStartId; + const observed = getProcessStartId(worker.descriptor.pid); + if (observed === undefined) { + return "unknown"; + } + return observed === worker.descriptor.processStartId ? "current" : "replaced"; } private async stopWorker( @@ -4638,12 +4644,14 @@ export class DaemonSupervisor { worker.client = undefined; } else if (directChild) { directChild.child.kill("SIGTERM"); - } else if (this.isWorkerProcessCurrent(worker)) { + } else if (this.workerProcessIdentity(worker) === "current") { signalProcessGroupOrProcess(worker.descriptor.pid, "SIGTERM"); } - // Identity-aware: a recycled pid is treated as gone and never signalled. - // kill(0) runs on every poll; the expensive identity check is throttled. - let identityVerdict = true; + // Identity-aware in both directions: a replaced pid counts as gone (never + // signal a recycled pid) while an unknown identity counts as alive (never + // clean up a possibly-live worker on a transient lookup failure). kill(0) + // runs on every poll; the expensive identity check is throttled. + let identityVerdict: "current" | "replaced" | "gone" | "unknown" = "current"; let identityCheckedAt = 0; const isWorkerProcessAlive = () => { if (directChild) { @@ -4655,9 +4663,9 @@ export class DaemonSupervisor { const now = Date.now(); if (now - identityCheckedAt >= LIVENESS_IDENTITY_RECHECK_MS) { identityCheckedAt = now; - identityVerdict = this.isWorkerProcessCurrent(worker); + identityVerdict = this.workerProcessIdentity(worker); } - return identityVerdict; + return identityVerdict !== "replaced" && identityVerdict !== "gone"; }; const gracefulDeadline = Date.now() + (force ? 500 : 2000); while (isWorkerProcessAlive() && Date.now() < gracefulDeadline) { @@ -4666,7 +4674,7 @@ export class DaemonSupervisor { if (force && isWorkerProcessAlive()) { if (directChild) { directChild.child.kill("SIGKILL"); - } else { + } else if (identityVerdict === "current") { signalProcessGroupOrProcess(worker.descriptor.pid, "SIGKILL"); } const forceDeadline = Date.now() + 1000; @@ -4741,7 +4749,11 @@ export class DaemonSupervisor { worker.stopRevision === stopRevision && worker.descriptor.stopRequestedAt !== undefined && worker.descriptor.pid === pid; + // A replaced pid counts as gone (never SIGKILL a recycled pid); an + // unobservable identity counts as alive (never clean up a possibly-live + // worker). kill(0) probes every poll; ps-backed checks are throttled. let stoppedVerdict = true; + let stoppedCanSignal = true; let stoppedCheckedAt = 0; const isStoppedProcessAlive = () => { if (!processIdExists(pid)) { @@ -4756,10 +4768,11 @@ export class DaemonSupervisor { stoppedVerdict = false; } else if (processStartId === undefined) { stoppedVerdict = true; + stoppedCanSignal = true; } else { - // Fail closed: an unobservable identity is treated as a different - // process so a recycled pid is never SIGKILLed. - stoppedVerdict = getProcessStartId(pid) === processStartId; + const observed = getProcessStartId(pid); + stoppedVerdict = observed !== processStartId ? observed === undefined : true; + stoppedCanSignal = observed === processStartId; } return stoppedVerdict; }; @@ -4772,7 +4785,7 @@ export class DaemonSupervisor { if (!isStoppedProcessAlive()) { break; } - if (!killed && Date.now() >= sigkillDeadline) { + if (!killed && stoppedCanSignal && Date.now() >= sigkillDeadline) { signalProcessGroupOrProcess(pid, "SIGKILL"); killed = true; } diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 4565d745c..145de7444 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1898,6 +1898,55 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("keeps waiting when process identity is transiently unobservable", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-unknown-identity-stop", + pid: 111_114, + processStartId: "proc:original", + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + stopWorker, + persistWorker: vi.fn(), + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const sessionLeaseModule = await import("../src/core/session-lease.js"); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + // Identity observation fails transiently (e.g. ps unavailable). + const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue(undefined); + try { + supervisor.scheduleWorkerStopFinalization(worker); + + await vi.advanceTimersByTimeAsync(20_000); + + // The possibly-live worker is neither signalled nor cleaned up. + expect(killSpy).not.toHaveBeenCalled(); + expect(stopWorker).not.toHaveBeenCalled(); + expect(worker.stopFinalization).toBeDefined(); + } finally { + existsSpy.mockRestore(); + aliveSpy.mockRestore(); + killSpy.mockRestore(); + startIdSpy.mockRestore(); + } + }); + it("retries finalization after a transient cleanup failure", async () => { vi.useFakeTimers(); const worker = { From fd5243d568470d069d3cd32eb8d9434d1224993c Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 7 Aug 2026 18:41:11 +0200 Subject: [PATCH 05/13] fix(coding-agent): abort stale stop cleanup when the worker is relaunched mid-await stopWorker can yield during archival while a retry rescinds the stop and relaunches the worker on the same registration. The cleanup tail now verifies the registered process is still the one it stopped before removing the registration or descriptor, so a relaunched worker is never orphaned by a stale stop invocation. --- .../src/modes/daemon/daemon-supervisor.ts | 10 ++++ .../test/daemon-supervisor-monitor.test.ts | 59 +++++++++++++++++++ 2 files changed, 69 insertions(+) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 4b092ccd3..fe8114a7e 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4596,6 +4596,14 @@ export class DaemonSupervisor { if (!recoveryCleanup) { worker.stopRevision++; } + // A retry can rescind this stop and relaunch the worker with a new + // process while we await below; never remove the successor's state. + const entryPid = worker.descriptor.pid; + const assertWorkerNotRelaunched = () => { + if (!directChild && worker.descriptor.pid !== entryPid) { + throw new Error(`Session worker ${worker.descriptor.workerId} was relaunched during stop`); + } + }; try { if (removeDescriptor) { this.persistWorkerStopTombstone(worker, archiveSession); @@ -4692,11 +4700,13 @@ export class DaemonSupervisor { if (directChild) { await directChild.closed; } + assertWorkerNotRelaunched(); if (removeDescriptor && worker.descriptor.archiveOnStop) { if (force) { this.reclaimStoppedWorkerCronLock(worker); } await this.finalizeArchivedWorkerStop(worker); + assertWorkerNotRelaunched(); } this.workers.delete(worker.descriptor.workerId); if (removeDescriptor) { diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 145de7444..30ba06915 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1898,6 +1898,65 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("aborts stale stop cleanup when the worker was relaunched during an await", async () => { + const worker = { + descriptor: { + workerId: "worker-relaunched-during-stop", + pid: 111_115, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString() as string | undefined, + archiveOnStop: true, + }, + client: undefined, + summaries: new Map(), + snapshotCache: new Map(), + transcriptCaches: new Map(), + snapshotGenerations: new Map(), + snapshotLoads: new Map(), + intentionalStop: true, + stopRevision: 0, + }; + const workers = new Map([[worker.descriptor.workerId, worker]]); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers, + shuttingDown: false, + persistWorker: vi.fn(), + persistWorkerStopTombstone: vi.fn(), + reclaimStoppedWorkerCronLock: vi.fn(), + // Archival yields, and a retry relaunches the worker meanwhile. + finalizeArchivedWorkerStop: vi.fn(async () => { + worker.descriptor.pid = 222_222; + worker.descriptor.stopRequestedAt = undefined; + }), + deleteWorkerDescriptor: vi.fn(), + syncAgentPeers: vi.fn(async () => {}), + broadcastHeartbeatsChanged: vi.fn(), + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as unknown as { + stopWorker( + target: object, + removeDescriptor: boolean, + force?: boolean, + archiveSession?: boolean, + ): Promise; + deleteWorkerDescriptor: ReturnType; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(false); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(false); + try { + await expect(supervisor.stopWorker(worker, true, true, true)).rejects.toThrow("was relaunched during stop"); + + // The relaunched worker's registration and descriptor must survive. + expect(workers.has(worker.descriptor.workerId)).toBe(true); + expect(supervisor.deleteWorkerDescriptor).not.toHaveBeenCalled(); + } finally { + existsSpy.mockRestore(); + aliveSpy.mockRestore(); + } + }); + it("keeps waiting when process identity is transiently unobservable", async () => { vi.useFakeTimers(); const worker = { From 1234a3130e1be007920029750021bec83d5d5b12 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 7 Aug 2026 20:32:11 +0200 Subject: [PATCH 06/13] fix(coding-agent): re-verify process identity at SIGKILL time The throttled identity cache can be up to 500ms old, long enough for a pid to be recycled. Both SIGKILL sites (stopWorker force escalation and the stop finalizer) now run a fresh identity check immediately before signalling; the cache remains only for read-only wait-loop polling. --- .../src/modes/daemon/daemon-supervisor.ts | 12 ++++- .../test/daemon-supervisor-monitor.test.ts | 52 +++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index fe8114a7e..720fcc8ed 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4682,7 +4682,9 @@ export class DaemonSupervisor { if (force && isWorkerProcessAlive()) { if (directChild) { directChild.child.kill("SIGKILL"); - } else if (identityVerdict === "current") { + } else if (this.workerProcessIdentity(worker) === "current") { + // Fresh, unthrottled check: the cached verdict may be up to 500ms + // old, long enough for the pid to be recycled. signalProcessGroupOrProcess(worker.descriptor.pid, "SIGKILL"); } const forceDeadline = Date.now() + 1000; @@ -4796,7 +4798,13 @@ export class DaemonSupervisor { break; } if (!killed && stoppedCanSignal && Date.now() >= sigkillDeadline) { - signalProcessGroupOrProcess(pid, "SIGKILL"); + // Fresh, unthrottled identity check right before signalling: the + // cached verdict may be up to 500ms old, long enough for the pid + // to be recycled by an unrelated process. + const observedNow = processStartId === undefined ? undefined : getProcessStartId(pid); + if (processStartId === undefined || observedNow === processStartId) { + signalProcessGroupOrProcess(pid, "SIGKILL"); + } killed = true; } await unrefDelay(STOP_FINALIZATION_RECHECK_MS); diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 30ba06915..89abb7565 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1957,6 +1957,58 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("re-verifies identity at SIGKILL time even within the throttle window", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-kill-window-recycle", + pid: 111_118, + processStartId: "proc:original", + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + const workers = new Map([[worker.descriptor.workerId, worker]]); + const stopWorker = vi.fn(async () => { + workers.delete(worker.descriptor.workerId); + }); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers, + shuttingDown: false, + stopWorker, + persistWorker: vi.fn(), + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const sessionLeaseModule = await import("../src/core/session-lease.js"); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + // The worker is current on the throttled polls, but the pid is recycled + // by the time the SIGKILL deadline arrives. + const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue("proc:original"); + try { + supervisor.scheduleWorkerStopFinalization(worker); + await vi.advanceTimersByTimeAsync(4900); + startIdSpy.mockReturnValue("proc:recycled"); + await vi.advanceTimersByTimeAsync(2000); + + // The fresh check at signal time sees the recycled pid and holds fire. + expect(killSpy).not.toHaveBeenCalled(); + } finally { + existsSpy.mockRestore(); + aliveSpy.mockRestore(); + killSpy.mockRestore(); + startIdSpy.mockRestore(); + } + }); + it("keeps waiting when process identity is transiently unobservable", async () => { vi.useFakeTimers(); const worker = { From 1a5a336e94a1e30068988d63e27bddd2a7b738ac Mon Sep 17 00:00:00 2001 From: Sebastian Date: Mon, 10 Aug 2026 15:59:24 +0200 Subject: [PATCH 07/13] fix(coding-agent): keep SIGKILL escalation armed through identity outages A transiently unobservable identity at the escalation deadline now skips that attempt without marking the kill done, so a later pass that re-verifies the original process still escalates instead of leaving a wedged worker registered forever. --- .../src/modes/daemon/daemon-supervisor.ts | 6 +- .../test/daemon-supervisor-monitor.test.ts | 61 +++++++++++++++++++ 2 files changed, 65 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 720fcc8ed..4471f3be0 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4800,12 +4800,14 @@ export class DaemonSupervisor { if (!killed && stoppedCanSignal && Date.now() >= sigkillDeadline) { // Fresh, unthrottled identity check right before signalling: the // cached verdict may be up to 500ms old, long enough for the pid - // to be recycled by an unrelated process. + // to be recycled by an unrelated process. A transiently + // unobservable identity skips this attempt but keeps escalation + // armed so a wedged worker is still killed on a later pass. const observedNow = processStartId === undefined ? undefined : getProcessStartId(pid); if (processStartId === undefined || observedNow === processStartId) { signalProcessGroupOrProcess(pid, "SIGKILL"); + killed = true; } - killed = true; } await unrefDelay(STOP_FINALIZATION_RECHECK_MS); } diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 89abb7565..25aee58ca 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -2009,6 +2009,67 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("retries SIGKILL after a transient identity outage at the deadline", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-kill-retry", + pid: 111_119, + processStartId: "proc:original", + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + const workers = new Map([[worker.descriptor.workerId, worker]]); + const stopWorker = vi.fn(async () => { + workers.delete(worker.descriptor.workerId); + }); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers, + shuttingDown: false, + stopWorker, + persistWorker: vi.fn(), + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const sessionLeaseModule = await import("../src/core/session-lease.js"); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); + let alive = true; + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockImplementation(() => alive); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation((_pid, signal) => { + if (signal === "SIGKILL") { + alive = false; + } + }); + // Identity observation is down when the SIGKILL deadline passes... + const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue(undefined); + try { + supervisor.scheduleWorkerStopFinalization(worker); + const finalization = worker.stopFinalization; + + await vi.advanceTimersByTimeAsync(8000); + expect(killSpy).not.toHaveBeenCalled(); + + // ...but once identity is observable again, escalation still fires. + startIdSpy.mockReturnValue("proc:original"); + await vi.advanceTimersByTimeAsync(5000); + await finalization; + expect(killSpy).toHaveBeenCalledWith(worker.descriptor.pid, "SIGKILL"); + expect(stopWorker).toHaveBeenCalled(); + } finally { + existsSpy.mockRestore(); + aliveSpy.mockRestore(); + killSpy.mockRestore(); + startIdSpy.mockRestore(); + } + }); + it("keeps waiting when process identity is transiently unobservable", async () => { vi.useFakeTimers(); const worker = { From 67dae028657eef96680b0e0292be5ba181627470 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Mon, 10 Aug 2026 16:13:05 +0200 Subject: [PATCH 08/13] fix(coding-agent): bind stopWorker to its entry process and detect rescinded stops All stopWorker polling and signalling now use the pid and start identity captured at entry, so a retry relaunching the worker mid-stop can never be SIGKILLed through the mutable descriptor. The cleanup guard also aborts when a removeDescriptor stop lost its tombstone, catching a rescission that lands before the successor pid does. --- .../src/modes/daemon/daemon-supervisor.ts | 57 +++++++++++------- .../test/daemon-supervisor-monitor.test.ts | 59 +++++++++++++++++++ 2 files changed, 94 insertions(+), 22 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 4471f3be0..20fda69eb 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4541,24 +4541,27 @@ export class DaemonSupervisor { } /** - * Verdict on whether the registered pid is still the process we launched. - * Callers must be conservative in both directions: signal a pid only on - * "current" (never SIGKILL a recycled pid), and clean up a registration - * only on "gone"/"replaced" (never orphan a live worker because a - * transient identity lookup failed). + * Verdict on whether a pid is still the process we launched. Callers must + * be conservative in both directions: signal a pid only on "current" + * (never SIGKILL a recycled pid), and clean up a registration only on + * "gone"/"replaced" (never orphan a live worker because a transient + * identity lookup failed). */ - private workerProcessIdentity(worker: ResidentWorker): "current" | "replaced" | "gone" | "unknown" { - if (!isProcessAlive(worker.descriptor.pid)) { + private processIdentity( + pid: number, + processStartId: string | undefined, + ): "current" | "replaced" | "gone" | "unknown" { + if (!isProcessAlive(pid)) { return "gone"; } - if (worker.descriptor.processStartId === undefined) { + if (processStartId === undefined) { return "current"; } - const observed = getProcessStartId(worker.descriptor.pid); + const observed = getProcessStartId(pid); if (observed === undefined) { return "unknown"; } - return observed === worker.descriptor.processStartId ? "current" : "replaced"; + return observed === processStartId ? "current" : "replaced"; } private async stopWorker( @@ -4596,11 +4599,21 @@ export class DaemonSupervisor { if (!recoveryCleanup) { worker.stopRevision++; } - // A retry can rescind this stop and relaunch the worker with a new - // process while we await below; never remove the successor's state. + // A retry can rescind this stop and relaunch the worker while we await + // below. Bind every liveness check and signal to the process this stop + // entered with, and abort cleanup once the stop no longer applies: the + // pid changed (relaunched) or a removeDescriptor stop lost its tombstone + // (rescinded, even before the successor pid lands). const entryPid = worker.descriptor.pid; - const assertWorkerNotRelaunched = () => { - if (!directChild && worker.descriptor.pid !== entryPid) { + const entryStartId = worker.descriptor.processStartId; + const assertStopStillApplies = () => { + if (directChild) { + return; + } + if ( + worker.descriptor.pid !== entryPid || + (removeDescriptor && worker.descriptor.stopRequestedAt === undefined) + ) { throw new Error(`Session worker ${worker.descriptor.workerId} was relaunched during stop`); } }; @@ -4652,8 +4665,8 @@ export class DaemonSupervisor { worker.client = undefined; } else if (directChild) { directChild.child.kill("SIGTERM"); - } else if (this.workerProcessIdentity(worker) === "current") { - signalProcessGroupOrProcess(worker.descriptor.pid, "SIGTERM"); + } else if (this.processIdentity(entryPid, entryStartId) === "current") { + signalProcessGroupOrProcess(entryPid, "SIGTERM"); } // Identity-aware in both directions: a replaced pid counts as gone (never // signal a recycled pid) while an unknown identity counts as alive (never @@ -4665,13 +4678,13 @@ export class DaemonSupervisor { if (directChild) { return directChild.child.exitCode === null && directChild.child.signalCode === null; } - if (!processIdExists(worker.descriptor.pid)) { + if (!processIdExists(entryPid)) { return false; } const now = Date.now(); if (now - identityCheckedAt >= LIVENESS_IDENTITY_RECHECK_MS) { identityCheckedAt = now; - identityVerdict = this.workerProcessIdentity(worker); + identityVerdict = this.processIdentity(entryPid, entryStartId); } return identityVerdict !== "replaced" && identityVerdict !== "gone"; }; @@ -4682,10 +4695,10 @@ export class DaemonSupervisor { if (force && isWorkerProcessAlive()) { if (directChild) { directChild.child.kill("SIGKILL"); - } else if (this.workerProcessIdentity(worker) === "current") { + } else if (this.processIdentity(entryPid, entryStartId) === "current") { // Fresh, unthrottled check: the cached verdict may be up to 500ms // old, long enough for the pid to be recycled. - signalProcessGroupOrProcess(worker.descriptor.pid, "SIGKILL"); + signalProcessGroupOrProcess(entryPid, "SIGKILL"); } const forceDeadline = Date.now() + 1000; while (isWorkerProcessAlive() && Date.now() < forceDeadline) { @@ -4702,13 +4715,13 @@ export class DaemonSupervisor { if (directChild) { await directChild.closed; } - assertWorkerNotRelaunched(); + assertStopStillApplies(); if (removeDescriptor && worker.descriptor.archiveOnStop) { if (force) { this.reclaimStoppedWorkerCronLock(worker); } await this.finalizeArchivedWorkerStop(worker); - assertWorkerNotRelaunched(); + assertStopStillApplies(); } this.workers.delete(worker.descriptor.workerId); if (removeDescriptor) { diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 25aee58ca..157acf14e 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1957,6 +1957,65 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("aborts stale stop cleanup when the stop is rescinded before the relaunch lands", async () => { + const worker = { + descriptor: { + workerId: "worker-rescinded-during-stop", + pid: 111_120, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString() as string | undefined, + archiveOnStop: true, + }, + client: undefined, + summaries: new Map(), + snapshotCache: new Map(), + transcriptCaches: new Map(), + snapshotGenerations: new Map(), + snapshotLoads: new Map(), + intentionalStop: true, + stopRevision: 0, + }; + const workers = new Map([[worker.descriptor.workerId, worker]]); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers, + shuttingDown: false, + persistWorker: vi.fn(), + persistWorkerStopTombstone: vi.fn(), + reclaimStoppedWorkerCronLock: vi.fn(), + // A retry rescinds the tombstone while archival yields, before + // recoverWorker has assigned the successor pid. + finalizeArchivedWorkerStop: vi.fn(async () => { + worker.descriptor.stopRequestedAt = undefined; + }), + deleteWorkerDescriptor: vi.fn(), + syncAgentPeers: vi.fn(async () => {}), + broadcastHeartbeatsChanged: vi.fn(), + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as unknown as { + stopWorker( + target: object, + removeDescriptor: boolean, + force?: boolean, + archiveSession?: boolean, + ): Promise; + deleteWorkerDescriptor: ReturnType; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(false); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(false); + try { + await expect(supervisor.stopWorker(worker, true, true, true)).rejects.toThrow("was relaunched during stop"); + + // The revived registration and descriptor must survive for recovery. + expect(workers.has(worker.descriptor.workerId)).toBe(true); + expect(supervisor.deleteWorkerDescriptor).not.toHaveBeenCalled(); + } finally { + existsSpy.mockRestore(); + aliveSpy.mockRestore(); + } + }); + it("re-verifies identity at SIGKILL time even within the throttle window", async () => { vi.useFakeTimers(); const worker = { From d849de71494faa5706b1c7213e2d0666d6f60898 Mon Sep 17 00:00:00 2001 From: Alex Zhang Date: Mon, 10 Aug 2026 23:59:30 -0400 Subject: [PATCH 09/13] fix(coding-agent): require worker identity before stop escalation --- .../src/modes/daemon/daemon-supervisor.ts | 7 ++- .../test/daemon-supervisor-monitor.test.ts | 46 +++++++++++++++++++ 2 files changed, 51 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 20fda69eb..0bbcb1a2a 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4778,7 +4778,7 @@ export class DaemonSupervisor { // unobservable identity counts as alive (never clean up a possibly-live // worker). kill(0) probes every poll; ps-backed checks are throttled. let stoppedVerdict = true; - let stoppedCanSignal = true; + let stoppedCanSignal = processStartId !== undefined; let stoppedCheckedAt = 0; const isStoppedProcessAlive = () => { if (!processIdExists(pid)) { @@ -4793,7 +4793,10 @@ export class DaemonSupervisor { stoppedVerdict = false; } else if (processStartId === undefined) { stoppedVerdict = true; - stoppedCanSignal = true; + // Without an identity captured while the original worker was known + // alive, this pid may now belong to an unrelated process. Keep + // waiting for it to disappear, but never escalate by pid alone. + stoppedCanSignal = false; } else { const observed = getProcessStartId(pid); stoppedVerdict = observed !== processStartId ? observed === undefined : true; diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 157acf14e..2ac3e39fc 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -2178,6 +2178,52 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("never SIGKILLs an identity-less worker pid", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-missing-identity", + pid: 111_121, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + intentionalStop: true, + stopRevision: 0, + stopFinalization: undefined as Promise | undefined, + }; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + stopWorker, + persistWorker: vi.fn(), + log: vi.fn(), + reportCleanupFailure: vi.fn(), + }) as { + scheduleWorkerStopFinalization(target: object): void; + }; + const childProcessModule = await import("../src/utils/child-process.js"); + const sessionLeaseModule = await import("../src/core/session-lease.js"); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue(undefined); + try { + supervisor.scheduleWorkerStopFinalization(worker); + + await vi.advanceTimersByTimeAsync(20_000); + + expect(killSpy).not.toHaveBeenCalled(); + expect(stopWorker).not.toHaveBeenCalled(); + expect(worker.stopFinalization).toBeDefined(); + } finally { + existsSpy.mockRestore(); + aliveSpy.mockRestore(); + killSpy.mockRestore(); + startIdSpy.mockRestore(); + } + }); + it("retries finalization after a transient cleanup failure", async () => { vi.useFakeTimers(); const worker = { From 56afc6b68ba655403f9109ee4656989940b193dc Mon Sep 17 00:00:00 2001 From: Alex Zhang Date: Tue, 11 Aug 2026 00:17:48 -0400 Subject: [PATCH 10/13] fix(coding-agent): reject unidentified worker pids Fixes #851 --- .../src/modes/daemon/daemon-supervisor.ts | 2 +- .../test/daemon-supervisor-monitor.test.ts | 58 ++++++++++++++++++- 2 files changed, 58 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 0bbcb1a2a..bed78ccb0 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4555,7 +4555,7 @@ export class DaemonSupervisor { return "gone"; } if (processStartId === undefined) { - return "current"; + return "unknown"; } const observed = getProcessStartId(pid); if (observed === undefined) { diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 2ac3e39fc..e9982072e 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -6,6 +6,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import type { AgentMessage } from "@earendil-works/pi-agent-core"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { getProcessStartId } from "../src/core/session-lease.js"; import type { DaemonSocketClient } from "../src/modes/daemon/active-session-state.js"; import { CommandRecoveryJournal } from "../src/modes/daemon/command-recovery-journal.js"; import { DaemonCatalogClient } from "../src/modes/daemon/daemon-catalog-process.js"; @@ -27,6 +28,7 @@ import { import { MutationDrainLatch } from "../src/modes/daemon/mutation-drain-latch.js"; import { WorkerRecoveryJournal } from "../src/modes/daemon/worker-recovery-journal.js"; import type { PrivateFrame } from "../src/modes/session-worker/private-framing.js"; +import * as childProcessModule from "../src/utils/child-process.js"; import { createDeferred } from "./suite/scheduling.js"; const workerLaunchTestState = vi.hoisted(() => ({ @@ -188,6 +190,7 @@ function createExistingLaunchWorker(root: string, descriptorDir: string) { version: 1 as const, workerId, pid: 999_999, + processStartId: undefined as string | undefined, socketPath: join(root, `${workerId}.sock`), recoveryJournalPath: join(descriptorDir, `${workerId}.recovery.jsonl`), orphanProcessJournalPath: join(descriptorDir, `${workerId}.orphans.jsonl`), @@ -898,6 +901,11 @@ describe("daemon worker supervisor monitoring", () => { ); await rollbackStarted; supervisor.shuttingDown = true; + workerLaunchTestState.forceMissingProcessStartId = false; + existing.descriptor.processStartId = getProcessStartId(existing.descriptor.pid); + if (existing.descriptor.processStartId === undefined) { + throw new Error("Could not identify launched worker before shutdown"); + } await supervisor.stopWorker(existing, true, true); releaseRollback(); @@ -1817,7 +1825,6 @@ describe("daemon worker supervisor monitoring", () => { }) as { scheduleWorkerStopFinalization(target: object): void; }; - const childProcessModule = await import("../src/utils/child-process.js"); const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); @@ -2016,6 +2023,55 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("never signals an identity-less worker pid during a forced stop", async () => { + vi.useFakeTimers(); + const worker = { + descriptor: { + workerId: "worker-force-missing-identity", + pid: 111_122, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + }, + client: undefined, + summaries: new Map(), + snapshotCache: new Map(), + transcriptCaches: new Map(), + snapshotGenerations: new Map(), + snapshotLoads: new Map(), + intentionalStop: true, + stopRevision: 0, + }; + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + workers: new Map([[worker.descriptor.workerId, worker]]), + shuttingDown: false, + persistWorker: vi.fn(), + persistWorkerStopTombstone: vi.fn(), + scheduleWorkerStopFinalization: vi.fn(), + syncAgentPeers: vi.fn(async () => {}), + broadcastHeartbeatsChanged: vi.fn(), + }) as unknown as { + stopWorker(target: object, removeDescriptor: boolean, force?: boolean): Promise; + scheduleWorkerStopFinalization: ReturnType; + }; + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + try { + const stopping = expect(supervisor.stopWorker(worker, true, true)).rejects.toThrow( + "did not stop after SIGKILL", + ); + await vi.advanceTimersByTimeAsync(2000); + + await stopping; + expect(killSpy).not.toHaveBeenCalled(); + expect(supervisor.scheduleWorkerStopFinalization).toHaveBeenCalledWith(worker); + } finally { + existsSpy.mockRestore(); + aliveSpy.mockRestore(); + killSpy.mockRestore(); + } + }); + it("re-verifies identity at SIGKILL time even within the throttle window", async () => { vi.useFakeTimers(); const worker = { From dc64f9b696594284e80ad1c1de3e65477e209cce Mon Sep 17 00:00:00 2001 From: Alex Zhang Date: Tue, 11 Aug 2026 00:23:02 -0400 Subject: [PATCH 11/13] fix(coding-agent): keep unknown worker identities untrusted Fixes #851 --- .../src/modes/daemon/daemon-supervisor.ts | 25 ++++++------------- .../test/daemon-supervisor-monitor.test.ts | 16 +++++++++--- 2 files changed, 19 insertions(+), 22 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index bed78ccb0..18240f230 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -4692,13 +4692,15 @@ export class DaemonSupervisor { while (isWorkerProcessAlive() && Date.now() < gracefulDeadline) { await delay(25); } + let sigkillSent = false; if (force && isWorkerProcessAlive()) { if (directChild) { - directChild.child.kill("SIGKILL"); + sigkillSent = directChild.child.kill("SIGKILL"); } else if (this.processIdentity(entryPid, entryStartId) === "current") { // Fresh, unthrottled check: the cached verdict may be up to 500ms // old, long enough for the pid to be recycled. signalProcessGroupOrProcess(entryPid, "SIGKILL"); + sigkillSent = true; } const forceDeadline = Date.now() + 1000; while (isWorkerProcessAlive() && Date.now() < forceDeadline) { @@ -4710,7 +4712,9 @@ export class DaemonSupervisor { if (removeDescriptor) { this.scheduleWorkerStopFinalization(worker); } - throw new Error(`Session worker ${worker.descriptor.workerId} did not stop${force ? " after SIGKILL" : ""}`); + throw new Error( + `Session worker ${worker.descriptor.workerId} did not stop${sigkillSent ? " after SIGKILL" : ""}`, + ); } if (directChild) { await directChild.closed; @@ -4752,22 +4756,7 @@ export class DaemonSupervisor { // the stop and relaunch with a new pid, and the OS can recycle the old // pid. The finalizer must never follow either successor. const pid = worker.descriptor.pid; - let processStartId = worker.descriptor.processStartId; - if (processStartId === undefined) { - // The stop timed out because the process was still alive moments ago, - // so an identity observed now can be trusted and recorded. It lets the - // eventual stopWorker call fail closed on a recycled pid too. - const observed = getProcessStartId(pid); - if (observed !== undefined && isProcessAlive(pid)) { - processStartId = observed; - worker.descriptor.processStartId = observed; - try { - this.persistWorker(worker); - } catch (error) { - this.reportCleanupFailure(`worker identity record ${worker.descriptor.workerId}`, error); - } - } - } + const processStartId = worker.descriptor.processStartId; const stopRevision = worker.stopRevision; const isStopGenerationCurrent = () => this.workers.get(worker.descriptor.workerId) === worker && diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index e9982072e..fe631e4bf 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -1755,10 +1755,15 @@ describe("daemon worker supervisor monitoring", () => { it("escalates a stuck stop to SIGKILL before finalizing", async () => { vi.useFakeTimers(); + const processStartId = getProcessStartId(process.pid); + if (processStartId === undefined) { + throw new Error("Could not identify test process"); + } const worker = { descriptor: { workerId: "worker-stuck-stop", pid: process.pid, + processStartId, rootActiveSessionId: "active-1", stopRequestedAt: new Date().toISOString(), archiveOnStop: true, @@ -2057,12 +2062,15 @@ describe("daemon worker supervisor monitoring", () => { const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); try { - const stopping = expect(supervisor.stopWorker(worker, true, true)).rejects.toThrow( - "did not stop after SIGKILL", + const stopping = supervisor.stopWorker(worker, true, true).then( + () => undefined, + (error: unknown) => error, ); await vi.advanceTimersByTimeAsync(2000); - await stopping; + const error = await stopping; + expect(error).toBeInstanceOf(Error); + expect((error as Error).message).toBe("Session worker worker-force-missing-identity did not stop"); expect(killSpy).not.toHaveBeenCalled(); expect(supervisor.scheduleWorkerStopFinalization).toHaveBeenCalledWith(worker); } finally { @@ -2263,7 +2271,7 @@ describe("daemon worker supervisor monitoring", () => { const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); - const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue(undefined); + const startIdSpy = vi.spyOn(sessionLeaseModule, "getProcessStartId").mockReturnValue("proc:unrelated"); try { supervisor.scheduleWorkerStopFinalization(worker); From 4a2f9c815f8f1aa8f31d17e63641adc254239c94 Mon Sep 17 00:00:00 2001 From: Alex Zhang Date: Tue, 11 Aug 2026 00:39:11 -0400 Subject: [PATCH 12/13] fix(coding-agent): finalize worker stops during shutdown Fixes #851 --- .../src/modes/daemon/daemon-supervisor.ts | 25 ++++-- .../test/daemon-supervisor-monitor.test.ts | 90 ++++++++++++++++++- 2 files changed, 107 insertions(+), 8 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 18240f230..d49646a0a 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -588,6 +588,7 @@ export class DaemonSupervisor { private ownership?: Awaited>; private cleanupPromise?: Promise; private shuttingDown = false; + private finalizeWorkerStopsDuringShutdown = false; private updateRestartPhase?: "draining" | "fencing" | "prepared"; private readonly mutationDrain = new MutationDrainLatch(); private readonly clients = new Set(); @@ -2382,9 +2383,6 @@ export class DaemonSupervisor { await this.assertRecoveryAllowed(); if (worker.descriptor.stopRequestedAt) { try { - // A tombstoned worker must not run long enough to elect another - // supervisor while its intentional stop is being adopted. - signalProcessGroupOrProcess(worker.descriptor.pid, "SIGKILL"); await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); this.log(`Completed intentional stop for worker ${worker.descriptor.workerId} during supervisor adoption`); } catch (error) { @@ -4795,7 +4793,8 @@ export class DaemonSupervisor { }; const sigkillDeadline = Date.now() + STOP_FINALIZATION_SIGKILL_GRACE_MS; let killed = false; - while (!this.shuttingDown) { + const shouldContinue = () => !this.shuttingDown || this.finalizeWorkerStopsDuringShutdown; + while (shouldContinue()) { if (!isStopGenerationCurrent()) { return; } @@ -4824,7 +4823,7 @@ export class DaemonSupervisor { this.workers.get(worker.descriptor.workerId) === worker && worker.descriptor.stopRequestedAt !== undefined && worker.descriptor.pid === pid; - while (!this.shuttingDown && isCleanupStillWanted()) { + while (shouldContinue() && isCleanupStillWanted()) { try { await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); this.log(`Finalized timed-out stop for worker ${worker.descriptor.workerId}`); @@ -5046,8 +5045,22 @@ export class DaemonSupervisor { cleanup(); } if (stopWorkers) { + this.finalizeWorkerStopsDuringShutdown = true; await Promise.all( - [...this.workers.values()].map((worker) => this.stopWorker(worker, true, forceWorkers, true)), + [...this.workers.values()].map(async (worker) => { + try { + await this.stopWorker(worker, true, forceWorkers, true); + } catch (error) { + const finalization = worker.stopFinalization; + if (!finalization) { + throw error; + } + await finalization; + if (this.workers.get(worker.descriptor.workerId) === worker) { + throw error; + } + } + }), ); if (!this.hasPersistedWorkerDescriptors()) { rmSync(this.supervisorConfigPath, { force: true }); diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index fe631e4bf..0a0516807 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -963,6 +963,62 @@ describe("daemon worker supervisor monitoring", () => { } }); + it("waits for timed-out worker finalization before completing shutdown", async () => { + const root = mkdtempSync(join(tmpdir(), "prime-supervisor-shutdown-finalization-test-")); + supervisorRegistryDirs.add(root); + const worker = { + descriptor: { workerId: "worker-shutdown-finalization" }, + stopFinalization: undefined as Promise | undefined, + }; + const workers = new Map([[worker.descriptor.workerId, worker]]); + let finishFinalization = () => {}; + const finalization = new Promise((resolve) => { + finishFinalization = resolve; + }).then(() => { + workers.delete(worker.descriptor.workerId); + }); + const stopError = new Error("worker stop timed out"); + const stopWorker = vi.fn(async () => { + worker.stopFinalization = finalization; + throw stopError; + }); + const exit = vi.spyOn(process, "exit").mockImplementation(((code?: string | number | null) => { + throw new Error(`exit ${code}`); + }) as typeof process.exit); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + shuttingDown: false, + finalizeWorkerStopsDuringShutdown: false, + signalCleanupHandlers: [], + workers, + clients: new Set(), + stopWorker, + hasPersistedWorkerDescriptors: vi.fn(() => true), + catalog: { stop: vi.fn(async () => undefined) }, + cleanupSocket: vi.fn(), + snapshotCacheRoot: join(root, "cache"), + log: vi.fn(), + }) as { + finalizeWorkerStopsDuringShutdown: boolean; + shutdown(exitCode: number, stopWorkers: boolean): Promise; + }; + + try { + const shutdown = supervisor.shutdown(0, true).then( + () => undefined, + (error: unknown) => error, + ); + await vi.waitFor(() => expect(stopWorker).toHaveBeenCalledOnce()); + expect(exit).not.toHaveBeenCalled(); + expect(supervisor.finalizeWorkerStopsDuringShutdown).toBe(true); + + finishFinalization(); + await expect(shutdown).resolves.toEqual(new Error("exit 0")); + expect(exit).toHaveBeenCalledWith(0); + } finally { + exit.mockRestore(); + } + }); + it("does not poll a healthy supervisor after the startup check", async () => { vi.useFakeTimers(); let resolveProbe: () => void = () => undefined; @@ -1707,6 +1763,35 @@ describe("daemon worker supervisor monitoring", () => { ]); }); + it("adopts a tombstoned worker through identity-aware stop handling", async () => { + const worker = { + descriptor: { + workerId: "worker-adopted-stop", + pid: 111_123, + processStartId: "proc:original", + stopRequestedAt: new Date().toISOString(), + archiveOnStop: true, + }, + }; + const stopWorker = vi.fn(async () => {}); + const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { + assertRecoveryAllowed: vi.fn(async () => undefined), + stopWorker, + log: vi.fn(), + }) as { + adoptOrRecoverWorker(target: object): Promise; + }; + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + try { + await supervisor.adoptOrRecoverWorker(worker); + + expect(killSpy).not.toHaveBeenCalled(); + expect(stopWorker).toHaveBeenCalledWith(worker, true, true, true); + } finally { + killSpy.mockRestore(); + } + }); + it("finalizes a timed-out stop once the worker process dies", async () => { vi.useFakeTimers(); const worker = { @@ -1753,7 +1838,7 @@ describe("daemon worker supervisor monitoring", () => { } }); - it("escalates a stuck stop to SIGKILL before finalizing", async () => { + it("escalates a stuck stop during shutdown before finalizing", async () => { vi.useFakeTimers(); const processStartId = getProcessStartId(process.pid); if (processStartId === undefined) { @@ -1776,7 +1861,8 @@ describe("daemon worker supervisor monitoring", () => { const stopWorker = vi.fn(async () => {}); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([[worker.descriptor.workerId, worker]]), - shuttingDown: false, + shuttingDown: true, + finalizeWorkerStopsDuringShutdown: true, stopWorker, persistWorker: vi.fn(), log: vi.fn(), From 0b0c43ce23c6f45b2f06b608bcb4d5cb95012549 Mon Sep 17 00:00:00 2001 From: Alex Zhang Date: Tue, 11 Aug 2026 00:49:45 -0400 Subject: [PATCH 13/13] fix(coding-agent): bound worker finalization during shutdown --- .../src/modes/daemon/daemon-supervisor.ts | 21 +++--- .../test/daemon-supervisor-monitor.test.ts | 68 +++++++++++-------- 2 files changed, 48 insertions(+), 41 deletions(-) diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index d49646a0a..eae3124b1 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -337,6 +337,8 @@ class SupervisorRecoveryCancelledError extends Error { class SnapshotLoadInvalidatedError extends Error {} +class WorkerStopTimeoutError extends Error {} + function isSupervisorGenerationStale(error: unknown): boolean { return ( typeof error === "object" && @@ -588,7 +590,6 @@ export class DaemonSupervisor { private ownership?: Awaited>; private cleanupPromise?: Promise; private shuttingDown = false; - private finalizeWorkerStopsDuringShutdown = false; private updateRestartPhase?: "draining" | "fencing" | "prepared"; private readonly mutationDrain = new MutationDrainLatch(); private readonly clients = new Set(); @@ -4710,7 +4711,7 @@ export class DaemonSupervisor { if (removeDescriptor) { this.scheduleWorkerStopFinalization(worker); } - throw new Error( + throw new WorkerStopTimeoutError( `Session worker ${worker.descriptor.workerId} did not stop${sigkillSent ? " after SIGKILL" : ""}`, ); } @@ -4793,8 +4794,7 @@ export class DaemonSupervisor { }; const sigkillDeadline = Date.now() + STOP_FINALIZATION_SIGKILL_GRACE_MS; let killed = false; - const shouldContinue = () => !this.shuttingDown || this.finalizeWorkerStopsDuringShutdown; - while (shouldContinue()) { + while (!this.shuttingDown) { if (!isStopGenerationCurrent()) { return; } @@ -4823,7 +4823,7 @@ export class DaemonSupervisor { this.workers.get(worker.descriptor.workerId) === worker && worker.descriptor.stopRequestedAt !== undefined && worker.descriptor.pid === pid; - while (shouldContinue() && isCleanupStillWanted()) { + while (!this.shuttingDown && isCleanupStillWanted()) { try { await this.stopWorker(worker, true, true, worker.descriptor.archiveOnStop === true); this.log(`Finalized timed-out stop for worker ${worker.descriptor.workerId}`); @@ -5045,20 +5045,17 @@ export class DaemonSupervisor { cleanup(); } if (stopWorkers) { - this.finalizeWorkerStopsDuringShutdown = true; await Promise.all( [...this.workers.values()].map(async (worker) => { try { await this.stopWorker(worker, true, forceWorkers, true); } catch (error) { - const finalization = worker.stopFinalization; - if (!finalization) { - throw error; - } - await finalization; - if (this.workers.get(worker.descriptor.workerId) === worker) { + if (!(error instanceof WorkerStopTimeoutError)) { throw error; } + this.log( + `Worker ${worker.descriptor.workerId} remains tombstoned for recovery after shutdown: ${error.message}`, + ); } }), ); diff --git a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts index 0a0516807..7748ef2ac 100644 --- a/packages/coding-agent/test/daemon-supervisor-monitor.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-monitor.test.ts @@ -963,59 +963,70 @@ describe("daemon worker supervisor monitoring", () => { } }); - it("waits for timed-out worker finalization before completing shutdown", async () => { + it("completes shutdown without awaiting an unsignalable worker finalizer", async () => { + vi.useFakeTimers(); const root = mkdtempSync(join(tmpdir(), "prime-supervisor-shutdown-finalization-test-")); supervisorRegistryDirs.add(root); const worker = { - descriptor: { workerId: "worker-shutdown-finalization" }, - stopFinalization: undefined as Promise | undefined, + descriptor: { + workerId: "worker-shutdown-finalization", + pid: 111_123, + rootActiveSessionId: "active-1", + stopRequestedAt: new Date().toISOString(), + archiveOnStop: true, + }, + client: undefined, + summaries: new Map(), + snapshotCache: new Map(), + transcriptCaches: new Map(), + snapshotGenerations: new Map(), + snapshotLoads: new Map(), + intentionalStop: true, + stopRevision: 1, + stopFinalization: new Promise(() => {}), }; const workers = new Map([[worker.descriptor.workerId, worker]]); - let finishFinalization = () => {}; - const finalization = new Promise((resolve) => { - finishFinalization = resolve; - }).then(() => { - workers.delete(worker.descriptor.workerId); - }); - const stopError = new Error("worker stop timed out"); - const stopWorker = vi.fn(async () => { - worker.stopFinalization = finalization; - throw stopError; - }); const exit = vi.spyOn(process, "exit").mockImplementation(((code?: string | number | null) => { throw new Error(`exit ${code}`); }) as typeof process.exit); + const killSpy = vi.spyOn(childProcessModule, "signalProcessGroupOrProcess").mockImplementation(() => {}); + const existsSpy = vi.spyOn(childProcessModule, "processIdExists").mockReturnValue(true); + const aliveSpy = vi.spyOn(childProcessModule, "isProcessAlive").mockReturnValue(true); + const catalogStop = vi.fn(async () => undefined); + const log = vi.fn(); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { shuttingDown: false, - finalizeWorkerStopsDuringShutdown: false, signalCleanupHandlers: [], workers, clients: new Set(), - stopWorker, + persistWorkerStopTombstone: vi.fn(), hasPersistedWorkerDescriptors: vi.fn(() => true), - catalog: { stop: vi.fn(async () => undefined) }, + catalog: { stop: catalogStop }, cleanupSocket: vi.fn(), snapshotCacheRoot: join(root, "cache"), - log: vi.fn(), + log, }) as { - finalizeWorkerStopsDuringShutdown: boolean; - shutdown(exitCode: number, stopWorkers: boolean): Promise; + shutdown(exitCode: number, stopWorkers: boolean, relaunch?: boolean, forceWorkers?: boolean): Promise; }; try { - const shutdown = supervisor.shutdown(0, true).then( + const shutdown = supervisor.shutdown(0, true, false, true).then( () => undefined, (error: unknown) => error, ); - await vi.waitFor(() => expect(stopWorker).toHaveBeenCalledOnce()); - expect(exit).not.toHaveBeenCalled(); - expect(supervisor.finalizeWorkerStopsDuringShutdown).toBe(true); - - finishFinalization(); + await vi.advanceTimersByTimeAsync(2000); await expect(shutdown).resolves.toEqual(new Error("exit 0")); + + expect(workers.has(worker.descriptor.workerId)).toBe(true); + expect(killSpy).not.toHaveBeenCalled(); + expect(catalogStop).toHaveBeenCalledOnce(); + expect(log).toHaveBeenCalledWith(expect.stringContaining("remains tombstoned for recovery")); expect(exit).toHaveBeenCalledWith(0); } finally { exit.mockRestore(); + killSpy.mockRestore(); + existsSpy.mockRestore(); + aliveSpy.mockRestore(); } }); @@ -1838,7 +1849,7 @@ describe("daemon worker supervisor monitoring", () => { } }); - it("escalates a stuck stop during shutdown before finalizing", async () => { + it("escalates a stuck stop to SIGKILL before finalizing", async () => { vi.useFakeTimers(); const processStartId = getProcessStartId(process.pid); if (processStartId === undefined) { @@ -1861,8 +1872,7 @@ describe("daemon worker supervisor monitoring", () => { const stopWorker = vi.fn(async () => {}); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([[worker.descriptor.workerId, worker]]), - shuttingDown: true, - finalizeWorkerStopsDuringShutdown: true, + shuttingDown: false, stopWorker, persistWorker: vi.fn(), log: vi.fn(),