diff --git a/packages/coding-agent/.changes/eng-5302.md b/packages/coding-agent/.changes/eng-5302.md new file mode 100644 index 0000000000..dea44dc3c1 --- /dev/null +++ b/packages/coding-agent/.changes/eng-5302.md @@ -0,0 +1 @@ +- Added `doctor` detection of daemons that lost their supervisor ownership record (status `ownership-lost`); `doctor --fix` now restarts only the affected daemon on the same socket and agent dir, preserving sessions (declining daemons whose state lives under a different agent dir), and ownership and socket-lock errors name the cause and the incumbent owner. diff --git a/packages/coding-agent/src/cli/daemon-ps-format.ts b/packages/coding-agent/src/cli/daemon-ps-format.ts index a2277841d6..cfe8db6515 100644 --- a/packages/coding-agent/src/cli/daemon-ps-format.ts +++ b/packages/coding-agent/src/cli/daemon-ps-format.ts @@ -20,9 +20,21 @@ export function formatDaemonListTable(daemons: readonly DaemonInfo[]): string { uptime: formatUptime(daemon.uptimeSeconds), })); const table = formatTable(["socket", "pid", "version", "status", "sessions", "uptime"], rows, formatDaemonCell); - return daemons.some((daemon) => daemon.isDefault) - ? `${table}\n\n${chalk.dim("* default background service")}` - : table; + const footers: string[] = []; + if (daemons.some((daemon) => daemon.isDefault)) { + footers.push(chalk.dim("* default background service")); + } + for (const daemon of daemons) { + if (daemon.status === "ownership-lost") { + footers.push( + chalk.dim( + `! ${daemon.socketPath}: supervisor lost its ownership record — ` + + 'run "prime-agent doctor --fix" to restart it (sessions are preserved)', + ), + ); + } + } + return footers.length > 0 ? `${table}\n\n${footers.join("\n")}` : table; } function formatDaemonCell(_row: DaemonRow, column: keyof DaemonRow, value: string): string { @@ -38,6 +50,7 @@ function colorStatus(status: DaemonStatus, value: string): string { return chalk.green(value); case "stale": return chalk.yellow(value); + case "ownership-lost": case "unreachable": return chalk.red(value); case "orphan-file": diff --git a/packages/coding-agent/src/cli/daemon-ps.ts b/packages/coding-agent/src/cli/daemon-ps.ts index 518329622d..0d4fc2069e 100644 --- a/packages/coding-agent/src/cli/daemon-ps.ts +++ b/packages/coding-agent/src/cli/daemon-ps.ts @@ -12,9 +12,15 @@ import { type DaemonRuntimeIdentity, } from "../modes/daemon/daemon-protocol.js"; import { defaultDaemonSocketDir, defaultDaemonSocketPath, normalizeSocketPath } from "../modes/daemon/daemon-socket.js"; -import { acquireDaemonShutdownAdmission } from "../modes/daemon/daemon-supervisor-ownership.js"; +import { supervisorStateDirMatches } from "../modes/daemon/daemon-supervisor.js"; +import { + acquireDaemonShutdownAdmission, + assertDaemonSupervisorOwnerCurrent, + waitForDaemonStartupFence, +} from "../modes/daemon/daemon-supervisor-ownership.js"; import type { DaemonWorkerDescriptor } from "../modes/daemon/daemon-worker-protocol.js"; import { signalProcessGroupOrProcess } from "../utils/child-process.js"; +import { ensureInteractiveDaemonRunning } from "./daemon-launch.js"; import { formatDaemonListTable } from "./daemon-ps-format.js"; import { promptYesNo } from "./daemon-stop-confirm.js"; @@ -36,7 +42,7 @@ import { promptYesNo } from "./daemon-stop-confirm.js"; * older build (a new protocol command would not). */ -export type DaemonStatus = "current" | "stale" | "unreachable" | "orphan-file"; +export type DaemonStatus = "current" | "stale" | "ownership-lost" | "unreachable" | "orphan-file"; export interface DiscoveredDaemonProcess { pid: number; @@ -63,8 +69,9 @@ export interface DaemonInfo { const STATUS_ORDER: Record = { current: 0, stale: 1, - unreachable: 2, - "orphan-file": 3, + "ownership-lost": 2, + unreachable: 3, + "orphan-file": 4, }; const SHUTDOWN_QUIET_PERIOD_MS = 1000; const SHUTDOWN_CONVERGENCE_TIMEOUT_MS = 10_000; @@ -233,12 +240,13 @@ function scanSocketDir(): string[] { return sockets; } -interface ProbeResult { +export interface ProbeResult { version?: string; protocolVersion?: number; schemaId?: string; runtime?: DaemonRuntimeIdentity; sessionCount?: number; + supervisorGeneration?: string; supervisorPid?: number; supervisorProcessStartId?: string; reachable: boolean; @@ -257,6 +265,7 @@ async function probeDaemon(socketPath: string): Promise { let protocolVersion: number | undefined; let schemaId: string | undefined; let runtime: DaemonRuntimeIdentity | undefined; + let supervisorGeneration: string | undefined; let supervisorPid: number | undefined; let supervisorProcessStartId: string | undefined; let greeted = false; @@ -266,6 +275,7 @@ async function probeDaemon(socketPath: string): Promise { protocolVersion = hello.protocol.version; schemaId = hello.schemaId; runtime = hello.runtime; + supervisorGeneration = hello.supervisorGeneration; supervisorPid = hello.supervisorPid; supervisorProcessStartId = hello.supervisorProcessStartId; greeted = true; @@ -290,6 +300,7 @@ async function probeDaemon(socketPath: string): Promise { schemaId, runtime, sessionCount, + supervisorGeneration, supervisorPid, supervisorProcessStartId, reachable: true, @@ -333,6 +344,48 @@ export function verifyHelloSupervisorPid( return pid; } +/** + * Cross-checks the probed hello identity against the on-disk ownership + * registry; no daemon commands (a wedged supervisor cannot serve them). Race- + * free without age heuristics: the record is written before listen() and + * released only after the socket closes, and the final pid+startId liveness + * check rejects daemons that exited or were replaced since the probe. + */ +export async function detectDaemonOwnershipLost( + socketPath: string, + probe: ProbeResult, + registryDir?: string, +): Promise { + if (!probe.reachable || !probe.supervisorGeneration) { + return false; + } + const pid = verifyHelloSupervisorPid(probe.supervisorPid, probe.supervisorProcessStartId); + if (pid === undefined) { + return false; + } + try { + await assertDaemonSupervisorOwnerCurrent( + { + generation: probe.supervisorGeneration, + pid, + ...(probe.supervisorProcessStartId ? { processStartId: probe.supervisorProcessStartId } : {}), + socketPath, + }, + undefined, + registryDir, + ); + return false; + } catch (error) { + // This result gates a kill: an unexpected registry error must never + // classify a healthy daemon as lost. + if ((error as { code?: unknown }).code !== "supervisor_generation_stale") { + return false; + } + // Flag only a live process: a stopping daemon releases its record late. + return verifyHelloSupervisorPid(pid, probe.supervisorProcessStartId) !== undefined; + } +} + /** Discover every daemon on the machine and probe each for version + session count. */ export async function discoverDaemons(): Promise { const processBySocket = new Map(); @@ -359,11 +412,17 @@ export async function discoverDaemons(): Promise { const probe = await probeDaemon(socketPath); const pid = proc?.pid ?? verifyHelloSupervisorPid(probe.supervisorPid, probe.supervisorProcessStartId); const hasTrackedWorkers = workerSockets.has(socketPath); - const status: DaemonStatus = probe.reachable + let status: DaemonStatus = probe.reachable ? classifyReachable(probe) : proc || hasTrackedWorkers ? "unreachable" : "orphan-file"; + // Stale daemons wedge the same way — an upgraded CLI probing a pre-upgrade + // supervisor is the incident-report case, so version mismatch must not + // mask ownership loss. + if ((status === "current" || status === "stale") && (await detectDaemonOwnershipLost(socketPath, probe))) { + status = "ownership-lost"; + } return { socketPath, pid, @@ -412,6 +471,7 @@ export async function runPs(json: boolean): Promise { export type ReapAction = | { kind: "remove-file"; daemon: DaemonInfo } | { kind: "kill"; daemon: DaemonInfo } + | { kind: "restart"; daemon: DaemonInfo } | { kind: "shutdown"; daemon: DaemonInfo } | { kind: "skip"; daemon: DaemonInfo; reason: string }; @@ -445,6 +505,10 @@ export function planReap(daemons: readonly DaemonInfo[], force: boolean): ReapAc if (daemon.status === "orphan-file") { return { kind: "remove-file", daemon }; } + // Wedged supervisors reject every command; only a restart repairs them. + if (daemon.status === "ownership-lost") { + return { kind: "restart", daemon }; + } if (daemon.isDefault) { return { kind: "skip", daemon, reason: "default background service" }; } @@ -489,7 +553,8 @@ const SHUTDOWN_ALL_ACTION_ORDER: Record = { shutdown: 0, "remove-file": 1, kill: 2, - skip: 3, + restart: 3, + skip: 4, }; export type ShutdownConfirmationPlan = "none" | "prompt" | "json-error" | "tty-error"; @@ -1067,6 +1132,111 @@ async function stopTrackedProcess( return !isProcessAlive(pid); } +export interface RepairHooks { + probe: (socketPath: string) => Promise; + detectLost: (socketPath: string, probe: ProbeResult) => Promise; + ownsSupervisorState: (socketPath: string, supervisorGeneration: string) => boolean; + acquireAdmission: () => Promise<{ assertOrRenew: () => Promise; release: () => Promise }>; + killDaemon: (pid: number, expectedProcessStartId: string | undefined) => Promise; + waitStartupFence: (socketPath: string) => Promise; + ensureDaemonRunning: (socketPath: string) => Promise; +} + +const defaultRepairHooks: RepairHooks = { + probe: probeDaemon, + detectLost: detectDaemonOwnershipLost, + ownsSupervisorState: (socketPath, supervisorGeneration) => + supervisorStateDirMatches(getAgentDir(), socketPath, supervisorGeneration), + acquireAdmission: acquireDaemonShutdownAdmission, + killDaemon: forceKillDaemon, + waitStartupFence: waitForDaemonStartupFence, + ensureDaemonRunning: ensureInteractiveDaemonRunning, +}; + +/** + * Kill the wedged supervisor and relaunch on the same socket, mirroring the + * update-restart coordinator's handoff: shutdown admission is held across + * kill, socket cleanup, and fence wait (no concurrent launch can bind the + * socket mid-stop), then released before relaunch since a successor cannot + * acquire ownership under it — if a client autostart wins that relaunch race, + * ensureDaemonRunning converges on the same healthy outcome. Repair first + * proves the daemon's state dir lives under this process's agent dir and + * declines otherwise: the relaunch inherits this environment, so a foreign + * agent dir's workers could not be re-adopted. + */ +export async function repairOwnershipLostDaemon( + daemon: DaemonInfo, + hooks: RepairHooks = defaultRepairHooks, +): Promise { + const { socketPath } = daemon; + // A replaced daemon must never be killed for its predecessor. + const probe = await hooks.probe(socketPath); + if (!probe.reachable) { + return { skipped: "no longer reachable; not restarting" }; + } + if (!(await hooks.detectLost(socketPath, probe))) { + return { skipped: "no longer ownership-lost; not restarting" }; + } + if (verifyHelloSupervisorPid(probe.supervisorPid, probe.supervisorProcessStartId) === undefined) { + return { skipped: "ownership-lost but no verified pid to restart" }; + } + if (!probe.supervisorGeneration || !hooks.ownsSupervisorState(socketPath, probe.supervisorGeneration)) { + return { + skipped: + "daemon belongs to a different agent dir; not restarting " + + `(rerun "${APP_NAME} doctor --fix" with that daemon's agent dir configured)`, + }; + } + const admission = await hooks.acquireAdmission(); + let pid: number | undefined; + let killedPid: number | undefined; + try { + // Acquiring the admission may have waited; only a pid verified from a + // fresh hello under it is ever signaled. + const recheck = await hooks.probe(socketPath); + pid = + recheck.reachable && recheck.supervisorGeneration === probe.supervisorGeneration + ? verifyHelloSupervisorPid(recheck.supervisorPid, recheck.supervisorProcessStartId) + : undefined; + if (pid === undefined || !(await hooks.detectLost(socketPath, recheck))) { + return { skipped: "no longer ownership-lost; not restarting" }; + } + let killed: boolean; + try { + await admission.assertOrRenew(); + killed = await hooks.killDaemon(pid, recheck.supervisorProcessStartId); + } catch (error) { + return { skipped: `could not stop wedged supervisor (pid ${pid}): ${String(error)}` }; + } + if (!killed) { + return { skipped: `wedged supervisor (pid ${pid}) did not exit after SIGKILL; not touching its socket` }; + } + killedPid = pid; + removeSocketFile(socketPath); + await hooks.waitStartupFence(socketPath); + await admission.assertOrRenew(); + } catch (error) { + return { skipped: relaunchFailureReason(killedPid, error) }; + } finally { + await admission.release(); + } + try { + await hooks.ensureDaemonRunning(socketPath); + } catch (error) { + return { skipped: relaunchFailureReason(killedPid, error) }; + } + return { + reaped: `restarted background service after ownership loss (killed pid ${pid}, relaunched on same socket)`, + }; +} + +function relaunchFailureReason(killedPid: number | undefined, error: unknown): string { + return killedPid === undefined + ? `could not repair ownership-lost supervisor: ${String(error)}` + : `killed wedged supervisor (pid ${killedPid}) but relaunch failed: ${String(error)}; ` + + "run any prime-agent command to autostart it"; +} + export async function runReap(json: boolean, force: boolean): Promise { const daemons = await discoverDaemons(); const reaped: Array<{ socketPath: string; action: string }> = []; @@ -1106,6 +1276,9 @@ export async function runReap(json: boolean, force: boolean): Promise { } break; } + case "restart": + apply(await repairOwnershipLostDaemon(action.daemon), socketPath, reaped, skipped); + break; case "shutdown": apply(await reapReachableDaemon(socketPath, pid), socketPath, reaped, skipped); break; @@ -1128,7 +1301,7 @@ export async function runReap(json: boolean, force: boolean): Promise { } } -type ReapOutcome = { reaped: string } | { skipped: string }; +export type ReapOutcome = { reaped: string } | { skipped: string }; function apply( outcome: ReapOutcome, @@ -1180,20 +1353,47 @@ function killDaemon(pid: number): void { } } -async function forceKillDaemon(pid: number): Promise { +/** + * SIGTERM, then SIGKILL; resolves true only once the process is confirmed gone. + * With an expected start id, each signal is fenced against PID reuse: a pid + * whose process identity changed since verification is treated as exited. + */ +async function forceKillDaemon(pid: number, expectedProcessStartId?: string): Promise { + // Only a PROVEN identity change counts as exited: an unreadable start id + // (transient ps failure) must neither signal a possibly-recycled pid nor + // report the supervisor gone while it may still hold the socket. + const identityChanged = () => { + if (expectedProcessStartId === undefined) return false; + const observed = getProcessStartId(pid); + return observed !== undefined && observed !== expectedProcessStartId; + }; + if (identityChanged()) { + return true; + } killDaemon(pid); - const deadline = Date.now() + 1000; + let deadline = Date.now() + 1000; while (Date.now() < deadline) { if (!isProcessAlive(pid)) { - return; + return true; } await delay(50); } + if (identityChanged()) { + return true; + } try { process.kill(pid, "SIGKILL"); } catch { // Process already exited between the liveness check and the kill. } + deadline = Date.now() + 2000; + while (isProcessAlive(pid) && Date.now() < deadline) { + if (identityChanged()) { + return true; + } + await delay(50); + } + return !isProcessAlive(pid) || identityChanged(); } function isProcessAlive(pid: number): boolean { diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts index 18fbd43f76..9e47c2c1b0 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor-ownership.ts @@ -101,17 +101,28 @@ class DaemonSupervisorAlreadyRunningError extends Error { } } +type DaemonSupervisorOwnershipLostReason = "released" | "record-missing" | "record-replaced"; + +const OWNERSHIP_LOST_REASON_TEXT: Record = { + released: "ownership was already released", + "record-missing": "owner record is missing on disk", + "record-replaced": "owner record on disk was replaced by another owner", +}; + class DaemonSupervisorOwnershipLostError extends Error { readonly code = "supervisor_generation_stale" as const; - constructor(generation: string, details: { socketPath?: string; registryDir?: string } = {}) { + constructor( + generation: string, + details: { socketPath?: string; registryDir?: string; reason: DaemonSupervisorOwnershipLostReason }, + ) { const context = [ details.socketPath ? `socket: ${details.socketPath}` : undefined, details.registryDir ? `registry: ${details.registryDir}` : undefined, ].filter((part) => part !== undefined); super( `Daemon supervisor generation ${generation} no longer owns its registry entry ` + - `(record on disk is missing or was replaced)${context.length > 0 ? `; ${context.join("; ")}` : ""}; ` + + `(${OWNERSHIP_LOST_REASON_TEXT[details.reason]})${context.length > 0 ? `; ${context.join("; ")}` : ""}; ` + "restart the daemon to recover — sessions are preserved", ); this.name = "DaemonSupervisorOwnershipLostError"; @@ -194,18 +205,22 @@ class DaemonSupervisorOwnership { async assertCurrent(): Promise { if (this.released) { - throw this.ownershipLostError(); + throw this.ownershipLostError("released"); } const current = readOwnerRecord(this.ownerDirectory); - if (!current || !sameOwnerRecord(current, this.record)) { - throw this.ownershipLostError(); + if (!current) { + throw this.ownershipLostError("record-missing"); + } + if (!sameOwnerRecord(current, this.record)) { + throw this.ownershipLostError("record-replaced"); } } - private ownershipLostError(): DaemonSupervisorOwnershipLostError { + private ownershipLostError(reason: DaemonSupervisorOwnershipLostReason): DaemonSupervisorOwnershipLostError { return new DaemonSupervisorOwnershipLostError(this.record.generation, { socketPath: this.record.socketPath, registryDir: this.registryDir, + reason, }); } @@ -481,22 +496,57 @@ export async function assertDaemonSupervisorOwnerCurrent( const current = readOwnerRecord(ownerDirectoryPath(registryDir, owner.generation)) ?? (legacyRegistryDir ? readOwnerRecord(ownerDirectoryPath(legacyRegistryDir, owner.generation)) : undefined); + if (!current) { + throw new DaemonSupervisorOwnershipLostError(owner.generation, { + socketPath: owner.socketPath, + registryDir, + reason: "record-missing", + }); + } if ( - !current || current.pid !== owner.pid || current.processStartId !== owner.processStartId || current.socketPath !== normalizeSocketPath(owner.socketPath) || !isProcessAlive(current.pid) ) { - throw new DaemonSupervisorOwnershipLostError(owner.generation, { socketPath: owner.socketPath, registryDir }); + throw new DaemonSupervisorOwnershipLostError(owner.generation, { + socketPath: owner.socketPath, + registryDir, + reason: "record-replaced", + }); } const fingerprint = ownerRecordFingerprint(current); if (fingerprint !== validatedFingerprint && !isProcessIdentityAlive(current)) { - throw new DaemonSupervisorOwnershipLostError(owner.generation, { socketPath: owner.socketPath, registryDir }); + throw new DaemonSupervisorOwnershipLostError(owner.generation, { + socketPath: owner.socketPath, + registryDir, + reason: "record-replaced", + }); } return fingerprint; } +/** Purely-local registry scan for the owner record of a socket, used for diagnostics. */ +export function findDaemonSupervisorOwnerForSocket( + socketPath: string, + registryDir: string = defaultDaemonSupervisorRegistryDir(), +): { pid: number; generation: string } | undefined { + const normalized = normalizeSocketPath(socketPath); + let directories: string[]; + try { + directories = listOwnerDirectories(registryDir); + } catch { + return undefined; + } + for (const directory of directories) { + const owner = readOwnerRecord(directory); + if (owner && owner.socketPath === normalized) { + return { pid: owner.pid, generation: owner.generation }; + } + } + return undefined; +} + export async function acquireDaemonShutdownAdmission(): Promise { const registryDir = defaultDaemonSupervisorRegistryDir(); const processStartId = getProcessStartId(process.pid); diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 5fa5f14376..8c4d5f2510 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -1,6 +1,15 @@ import { type ChildProcess, spawn } from "node:child_process"; import { createHash, randomBytes, randomUUID } from "node:crypto"; -import { chmodSync, mkdirSync, readdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs"; +import { + chmodSync, + existsSync, + mkdirSync, + readdirSync, + readFileSync, + renameSync, + rmSync, + writeFileSync, +} from "node:fs"; import { createServer, type Server, type Socket } from "node:net"; import { dirname, isAbsolute, join, resolve } from "node:path"; import { Writable } from "node:stream"; @@ -105,6 +114,7 @@ import { } from "./daemon-socket.js"; import { acquireDaemonSupervisorOwnership, + findDaemonSupervisorOwnerForSocket, isDaemonShutdownAdmissionActive, waitForDaemonStartupFence, } from "./daemon-supervisor-ownership.js"; @@ -528,6 +538,21 @@ function defaultWorkerDescriptorDir(agentDir: string, socketPath: string): strin return join(agentDir, "daemon-workers", descriptorKey(socketPath)); } +/** Per-generation supervisor state dir under agentDir; exists from start() until shutdown. */ +function supervisorStateDir(agentDir: string, socketPath: string, supervisorGeneration: string): string { + return join(defaultWorkerDescriptorDir(agentDir, socketPath), "snapshot-cache", supervisorGeneration); +} + +/** + * Whether the supervisor generation keeps its state under agentDir: start() + * creates snapshot-cache/ in the descriptor dir and only shutdown + * removes it, so a match proves a relaunch under agentDir re-adopts the same + * workers. + */ +export function supervisorStateDirMatches(agentDir: string, socketPath: string, supervisorGeneration: string): boolean { + return existsSync(supervisorStateDir(agentDir, socketPath, supervisorGeneration)); +} + export function idleEvictionSweepIntervalMs(idleEvictionMinutes: IdleEvictionMinutes): number { if (idleEvictionMinutes === "off") return IDLE_EVICTION_MAX_SWEEP_INTERVAL_MS; return Math.max( @@ -678,7 +703,19 @@ export class DaemonSupervisor { if (!agentDir) { throw new Error("Daemon supervisor config is missing agentDir"); } - this.socketLease = await acquireDaemonSocketPathLease(this.socketPath); + try { + this.socketLease = await acquireDaemonSocketPathLease(this.socketPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ELOCKED") { + throw error; + } + const owner = findDaemonSupervisorOwnerForSocket(this.socketPath); + throw new Error( + `Daemon socket ${this.socketPath} is locked by another supervisor` + + `${owner ? ` (pid ${owner.pid}, generation ${owner.generation})` : ""}; ` + + "a concurrent daemon launch won the socket", + ); + } await waitForDaemonStartupFence(this.socketPath); this.ownership = await acquireDaemonSupervisorOwnership({ socketPath: this.socketPath, diff --git a/packages/coding-agent/test/daemon-ps-format.test.ts b/packages/coding-agent/test/daemon-ps-format.test.ts index 1b3108b6ff..3a26e3cc75 100644 --- a/packages/coding-agent/test/daemon-ps-format.test.ts +++ b/packages/coding-agent/test/daemon-ps-format.test.ts @@ -47,4 +47,21 @@ describe("formatDaemonListTable", () => { expect(table).toContain("orphan-file"); expect(table).toContain("2h"); }); + + it("appends a repair hint for ownership-lost daemons", () => { + const table = stripAnsi( + formatDaemonListTable([ + { + socketPath: "/tmp/lost.sock", + pid: 42, + status: "ownership-lost", + isDefault: false, + }, + ]), + ); + expect(table).toContain("ownership-lost"); + expect(table).toContain( + '! /tmp/lost.sock: supervisor lost its ownership record — run "prime-agent doctor --fix"', + ); + }); }); diff --git a/packages/coding-agent/test/daemon-ps.test.ts b/packages/coding-agent/test/daemon-ps.test.ts index 6233a9353e..e8f69b69b2 100644 --- a/packages/coding-agent/test/daemon-ps.test.ts +++ b/packages/coding-agent/test/daemon-ps.test.ts @@ -1,10 +1,15 @@ +import { createHash } from "node:crypto"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; import { join } from "node:path"; -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it } from "vitest"; import { type DaemonInfo, + detectDaemonOwnershipLost, evaluateShutdownQuietPeriod, isWorkerSocketPath, mergeDiscoveredDaemonProcesses, + type ProbeResult, parseLsofListeners, parsePrimeAgentProcessIds, parsePsEtimes, @@ -12,11 +17,15 @@ import { planReap, planShutdownAll, planShutdownConfirmation, + type RepairHooks, + repairOwnershipLostDaemon, sortDaemons, verifyHelloSupervisorPid, } from "../src/cli/daemon-ps.js"; import { getProcessStartId } from "../src/core/session-lease.js"; import { defaultDaemonSocketDir } from "../src/modes/daemon/daemon-socket.js"; +import { supervisorStateDirMatches } from "../src/modes/daemon/daemon-supervisor.js"; +import { acquireDaemonSupervisorOwnership } from "../src/modes/daemon/daemon-supervisor-ownership.js"; describe("worker socket classification", () => { it.runIf(process.platform !== "win32")("recognizes only worker sockets in the default service directory", () => { @@ -261,6 +270,290 @@ describe("planShutdownConfirmation", () => { }); }); +describe("planReap ownership-lost", () => { + it("restarts an ownership-lost daemon even when it is the default", () => { + const plan = planReap( + [ + makeDaemon({ socketPath: "/tmp/default.sock", status: "ownership-lost", isDefault: true, pid: 11 }), + makeDaemon({ socketPath: "/tmp/other.sock", status: "ownership-lost", pid: 12 }), + ], + false, + ); + expect(plan.map((action) => action.kind)).toEqual(["restart", "restart"]); + }); +}); + +const cleanupDirs: string[] = []; + +afterEach(() => { + while (cleanupDirs.length > 0) { + const dir = cleanupDirs.pop(); + if (dir) rmSync(dir, { recursive: true, force: true }); + } +}); + +async function acquireTestOwnership(generation: string): Promise<{ + registryDir: string; + socketPath: string; + probe: ProbeResult; + ownerJsonPath: string; + ownerDir: string; +}> { + const root = mkdtempSync(join(tmpdir(), "doctor-ownership-")); + cleanupDirs.push(root); + const registryDir = join(root, "registry"); + const socketPath = join(root, "daemon.sock"); + const ownership = await acquireDaemonSupervisorOwnership({ + agentDir: join(root, "agent"), + appVersion: "test", + descriptorDir: join(root, "workers"), + generation, + registryDir, + socketPath, + }); + const record = ownership.record; + const probe: ProbeResult = { + reachable: true, + supervisorGeneration: record.generation, + supervisorPid: record.pid, + ...(record.processStartId ? { supervisorProcessStartId: record.processStartId } : {}), + }; + const ownerDir = join(registryDir, `${generation}.owner`); + return { registryDir, socketPath, probe, ownerJsonPath: join(ownerDir, "owner.json"), ownerDir }; +} + +describe("detectDaemonOwnershipLost", () => { + it("flags a stale-version supervisor whose owner record was deleted", async () => { + // The incident case: the CLI was upgraded because the daemon wedged, so the + // wedged supervisor answers with a pre-upgrade version. Version mismatch + // must not mask ownership loss. + const paths = await acquireTestOwnership("stale-lost-owner"); + rmSync(paths.ownerDir, { recursive: true, force: true }); + const probe: ProbeResult = { ...paths.probe, version: "0.0.1-old" }; + await expect(detectDaemonOwnershipLost(paths.socketPath, probe, paths.registryDir)).resolves.toBe(true); + }); + + it("does not flag a supervisor whose owner record is current", async () => { + const paths = await acquireTestOwnership("healthy-owner"); + await expect(detectDaemonOwnershipLost(paths.socketPath, paths.probe, paths.registryDir)).resolves.toBe(false); + }); + + it("flags a live supervisor whose owner record was deleted", async () => { + const paths = await acquireTestOwnership("deleted-owner"); + rmSync(paths.ownerDir, { recursive: true, force: true }); + await expect(detectDaemonOwnershipLost(paths.socketPath, paths.probe, paths.registryDir)).resolves.toBe(true); + }); + + it("flags a live supervisor whose owner record was replaced", async () => { + const paths = await acquireTestOwnership("replaced-owner"); + const record = JSON.parse(readFileSync(paths.ownerJsonPath, "utf8")) as { pid: number }; + record.pid = record.pid + 1; + writeFileSync(paths.ownerJsonPath, `${JSON.stringify(record, null, 2)}\n`); + await expect(detectDaemonOwnershipLost(paths.socketPath, paths.probe, paths.registryDir)).resolves.toBe(true); + }); + + it("never flags a hello without a supervisor generation", async () => { + // A starting daemon has no generation in its hello yet and is skipped. + const paths = await acquireTestOwnership("young-owner"); + rmSync(paths.ownerDir, { recursive: true, force: true }); + const probe: ProbeResult = { ...paths.probe }; + delete probe.supervisorGeneration; + await expect(detectDaemonOwnershipLost(paths.socketPath, probe, paths.registryDir)).resolves.toBe(false); + }); + + it("never flags a supervisor whose process identity no longer matches", async () => { + // Clean-shutdown race: a dead/replaced process identity means no flag. + const paths = await acquireTestOwnership("stopping-owner"); + rmSync(paths.ownerDir, { recursive: true, force: true }); + const probe: ProbeResult = { ...paths.probe, supervisorProcessStartId: "stale-start-id" }; + await expect(detectDaemonOwnershipLost(paths.socketPath, probe, paths.registryDir)).resolves.toBe(false); + }); +}); + +describe("supervisorStateDirMatches", () => { + it("matches only the generation whose snapshot-cache dir exists under the agent dir", () => { + const agentDir = mkdtempSync(join(tmpdir(), "doctor-agent-dir-")); + cleanupDirs.push(agentDir); + const socketPath = "/tmp/state-dir.sock"; + // Created by DaemonSupervisor.start() before it ever listens. + // Mirrors defaultWorkerDescriptorDir's key rule; the positive assertion below + // fails if production's path derivation ever drifts from this. + const stateDir = join( + agentDir, + "daemon-workers", + createHash("sha256").update(socketPath).digest("hex").slice(0, 12), + "snapshot-cache", + "gen-a", + ); + mkdirSync(stateDir, { recursive: true }); + expect(supervisorStateDirMatches(agentDir, socketPath, "gen-a")).toBe(true); + expect(supervisorStateDirMatches(agentDir, socketPath, "gen-b")).toBe(false); + expect(supervisorStateDirMatches(agentDir, "/tmp/other.sock", "gen-a")).toBe(false); + }); +}); + +describe("repairOwnershipLostDaemon", () => { + function makeLostDaemon(): DaemonInfo { + return makeDaemon({ socketPath: "/tmp/lost.sock", status: "ownership-lost", pid: process.pid }); + } + + function makeHooks(overrides: Partial = {}): { hooks: RepairHooks; calls: string[] } { + const calls: string[] = []; + const probe: ProbeResult = { + reachable: true, + supervisorGeneration: "lost-generation", + supervisorPid: process.pid, + ...(getProcessStartId(process.pid) ? { supervisorProcessStartId: getProcessStartId(process.pid) } : {}), + }; + const hooks: RepairHooks = { + probe: async (socketPath) => { + calls.push(`probe:${socketPath}`); + return probe; + }, + detectLost: async () => { + calls.push("detectLost"); + return true; + }, + ownsSupervisorState: (socketPath, supervisorGeneration) => { + calls.push(`ownsState:${socketPath}:${supervisorGeneration}`); + return true; + }, + acquireAdmission: async () => { + calls.push("acquireAdmission"); + let released = false; + return { + assertOrRenew: async () => { + calls.push("assertAdmission"); + }, + release: async () => { + if (!released) { + released = true; + calls.push("releaseAdmission"); + } + }, + }; + }, + killDaemon: async (pid, expectedProcessStartId) => { + calls.push(`kill:${pid}:${expectedProcessStartId ?? "no-start-id"}`); + return true; + }, + waitStartupFence: async (socketPath) => { + calls.push(`fence:${socketPath}`); + }, + ensureDaemonRunning: async (socketPath) => { + calls.push(`ensure:${socketPath}`); + }, + ...overrides, + }; + return { hooks, calls }; + } + + it("holds admission across kill, cleanup, and fence wait, releasing only right before the relaunch", async () => { + const { hooks, calls } = makeHooks(); + const outcome = await repairOwnershipLostDaemon(makeLostDaemon(), hooks); + expect(outcome).toEqual({ + reaped: `restarted background service after ownership loss (killed pid ${process.pid}, relaunched on same socket)`, + }); + expect(calls).toEqual([ + "probe:/tmp/lost.sock", + "detectLost", + "ownsState:/tmp/lost.sock:lost-generation", + "acquireAdmission", + "probe:/tmp/lost.sock", + "detectLost", + "assertAdmission", + // The start id from the fresh under-admission probe fences the kill + // against PID reuse. + `kill:${process.pid}:${getProcessStartId(process.pid) ?? "no-start-id"}`, + "fence:/tmp/lost.sock", + "assertAdmission", + "releaseAdmission", + "ensure:/tmp/lost.sock", + ]); + }); + + it("skips without killing when the socket was taken over while waiting for admission", async () => { + const probes: ProbeResult[] = [ + { + reachable: true, + supervisorGeneration: "lost-generation", + supervisorPid: process.pid, + ...(getProcessStartId(process.pid) ? { supervisorProcessStartId: getProcessStartId(process.pid) } : {}), + }, + { + reachable: true, + supervisorGeneration: "replacement-generation", + supervisorPid: process.pid, + ...(getProcessStartId(process.pid) ? { supervisorProcessStartId: getProcessStartId(process.pid) } : {}), + }, + ]; + const { hooks, calls } = makeHooks({ + probe: async () => { + const next = probes.shift(); + if (!next) throw new Error("unexpected probe"); + return next; + }, + }); + const outcome = await repairOwnershipLostDaemon(makeLostDaemon(), hooks); + expect(outcome).toEqual({ skipped: "no longer ownership-lost; not restarting" }); + expect(calls.some((call) => call.startsWith("kill") || call.startsWith("ensure"))).toBe(false); + expect(calls).toContain("releaseAdmission"); + }); + + it("does not unlink or relaunch when the wedged supervisor survives SIGKILL", async () => { + const { hooks, calls } = makeHooks({ + killDaemon: async () => false, + }); + const outcome = await repairOwnershipLostDaemon(makeLostDaemon(), hooks); + expect("skipped" in outcome && outcome.skipped).toContain("did not exit after SIGKILL"); + expect(calls.some((call) => call.startsWith("fence") || call.startsWith("ensure"))).toBe(false); + expect(calls).toContain("releaseAdmission"); + }); + + it("declines to restart a daemon whose state lives under a different agent dir", async () => { + // Killing a foreign daemon would silently drop its workers on relaunch. + const { hooks, calls } = makeHooks({ + ownsSupervisorState: () => false, + }); + const outcome = await repairOwnershipLostDaemon(makeLostDaemon(), hooks); + expect("skipped" in outcome && outcome.skipped).toContain("different agent dir"); + expect(calls.some((call) => call.startsWith("kill") || call.startsWith("ensure"))).toBe(false); + }); + + it("does not kill a daemon that recovered between discovery and repair", async () => { + const { hooks, calls } = makeHooks({ + detectLost: async () => false, + }); + const outcome = await repairOwnershipLostDaemon(makeLostDaemon(), hooks); + expect(outcome).toEqual({ skipped: "no longer ownership-lost; not restarting" }); + expect(calls.some((call) => call.startsWith("kill") || call.startsWith("ensure"))).toBe(false); + }); + + it("reports a kill failure without attempting a relaunch", async () => { + const { hooks, calls } = makeHooks({ + killDaemon: async () => { + throw new Error("kill refused"); + }, + }); + const outcome = await repairOwnershipLostDaemon(makeLostDaemon(), hooks); + expect("skipped" in outcome && outcome.skipped).toContain("kill refused"); + expect(calls.some((call) => call.startsWith("ensure"))).toBe(false); + expect(calls).toContain("releaseAdmission"); + }); + + it("names the killed pid and the autostart fallback when the relaunch fails", async () => { + const { hooks } = makeHooks({ + ensureDaemonRunning: async () => { + throw new Error("spawn failed"); + }, + }); + const outcome = await repairOwnershipLostDaemon(makeLostDaemon(), hooks); + expect("skipped" in outcome && outcome.skipped).toContain(`killed wedged supervisor (pid ${process.pid})`); + expect("skipped" in outcome && outcome.skipped).toContain("spawn failed"); + expect("skipped" in outcome && outcome.skipped).toContain("autostart"); + }); +}); + function makeDaemon(options: Partial & { socketPath: string; status: DaemonInfo["status"] }): DaemonInfo { return { isDefault: false, diff --git a/packages/coding-agent/test/daemon-socket.test.ts b/packages/coding-agent/test/daemon-socket.test.ts index c266906118..92f40295dc 100644 --- a/packages/coding-agent/test/daemon-socket.test.ts +++ b/packages/coding-agent/test/daemon-socket.test.ts @@ -254,3 +254,21 @@ describe("defaultDaemonSocketPath", () => { } }); }); + +describe("proper-lockfile contention error shape", () => { + it.runIf(process.platform !== "win32")("reports ELOCKED on a held lock", async () => { + // The daemon supervisor's socket-lease error wrapping keys off + // error.code === "ELOCKED"; this pins that library contract. + const dir = mkdtempSync(join(tmpdir(), "prime-lockfile-shape-")); + const target = join(dir, "daemon.sock"); + const release = await lockfile.lock(target, { realpath: false }); + try { + await expect(lockfile.lock(target, { realpath: false, retries: 0 })).rejects.toMatchObject({ + code: "ELOCKED", + }); + } finally { + await release(); + rmSync(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/coding-agent/test/daemon-supervisor-ownership.test.ts b/packages/coding-agent/test/daemon-supervisor-ownership.test.ts index c89e5c03f5..805cc77abf 100644 --- a/packages/coding-agent/test/daemon-supervisor-ownership.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-ownership.test.ts @@ -8,6 +8,7 @@ import { acquireDaemonShutdownAdmission, acquireDaemonSupervisorOwnership, assertDaemonSupervisorOwnerCurrent, + findDaemonSupervisorOwnerForSocket, persistDaemonStartupFenceFromOwner, } from "../src/modes/daemon/daemon-supervisor-ownership.js"; @@ -162,6 +163,7 @@ describe("daemon supervisor ownership registry", () => { await expect(mismatched.assertCurrent()).rejects.toMatchObject({ code: "supervisor_generation_stale", name: "DaemonSupervisorOwnershipLostError", + message: expect.stringContaining("owner record on disk was replaced by another owner") as string, }); expect(readJson(mismatchedPath).token).toBe("successor-token"); await mismatched.release(); @@ -170,9 +172,27 @@ describe("daemon supervisor ownership registry", () => { const reaped = await acquire(paths, "reaped-owner"); const reapedDir = ownerDir(paths, "reaped-owner"); rmSync(reapedDir, { recursive: true, force: true }); - await expect(reaped.assertCurrent()).rejects.toMatchObject({ code: "supervisor_generation_stale" }); + await expect(reaped.assertCurrent()).rejects.toMatchObject({ + code: "supervisor_generation_stale", + message: expect.stringContaining("owner record is missing on disk") as string, + }); expect(existsSync(reapedDir)).toBe(false); - await reaped.release(); + await expect(reaped.release()).resolves.toBeUndefined(); + await expect(reaped.assertCurrent()).rejects.toMatchObject({ + message: expect.stringContaining("ownership was already released") as string, + }); + }); + + it("finds the owner record for a socket by purely local reads", async () => { + const paths = createPaths(); + expect(findDaemonSupervisorOwnerForSocket(paths.socketPath, paths.registryDir)).toBeUndefined(); + const ownership = await acquire(paths, "socket-owner"); + expect(findDaemonSupervisorOwnerForSocket(paths.socketPath, paths.registryDir)).toMatchObject({ + pid: process.pid, + generation: "socket-owner", + }); + expect(findDaemonSupervisorOwnerForSocket(join(paths.root, "other.sock"), paths.registryDir)).toBeUndefined(); + await ownership.release(); }); it("does not resurrect the shutdown admission when release overtakes an in-flight renew", async () => {