Something went wrong. Try again.
fork of prime agent with changes i want but are not upstreamable probqbly (and some fixes)
Something went wrong. Try again.
45 kB · 1376 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377import { createHash } from "node:crypto";import { existsSync, lstatSync, readdirSync, readFileSync, rmSync, unlinkSync } from "node:fs";import { basename, dirname, join, resolve } from "node:path";import chalk from "chalk";import { APP_NAME, getAgentDir, VERSION } from "../config.js";import { isOrphanProcessIdentityCurrent, killOrphanProcess, readActiveOrphanProcesses, shouldReapOrphanProcess,} from "../core/orphan-process-journal.js";import { canonicalSessionPath, getProcessStartId } from "../core/session-lease.js";import { DaemonClient } from "../modes/daemon/daemon-client.js";import { DAEMON_PROTOCOL_VERSION, DAEMON_SCHEMA_ID, type DaemonRuntimeIdentity,} from "../modes/daemon/daemon-protocol.js";import { defaultDaemonSocketDir, defaultDaemonSocketPath, normalizeSocketPath } from "../modes/daemon/daemon-socket.js";import { createDaemonStateRootMatcher } from "../modes/daemon/daemon-state-root.js";import { acquireDaemonShutdownAdmission } from "../modes/daemon/daemon-supervisor-ownership.js";import type { DaemonWorkerDescriptor } from "../modes/daemon/daemon-worker-protocol.js";import { isProcessAlive, processGroupHasLiveMember, processIdExists, signalProcessGroupIfHeld, spawnSyncHidden,} from "../utils/child-process.js";import { formatDaemonListTable } from "./daemon-ps-format.js";import { promptYesNo } from "./daemon-stop-confirm.js";
/** * `daemon ps` discovers every prime-agent daemon in *our state root*, not just * the one on a single socket. Discovery has two sources merged by socket path: * * 1. The OS list of listening unix sockets owned by a prime-agent process * (`ss -lxp` on Linux, `lsof` on macOS). Daemons set process.title to * APP_NAME and carry nothing useful in argv, so the socket→pid mapping the * kernel keeps is the only reliable way to find daemons on arbitrary * `--daemon-socket` paths. This is the same data as `ss -lxp | grep * prime-agent`, just parsed. The kernel list is machine-wide, so it is * filtered to our state root (see createDaemonStateRootMatcher): a daemon * started under a different HOME or agent dir is another root's business, * and stopping it from here would kill unrelated live sessions. * 2. A sweep of the default socket dir, which catches orphaned socket *files* * left behind by daemons that are no longer running. * * Each discovered socket is then probed with the existing daemon_hello + list * primitives, so introspection works even against stale daemons running an * older build (a new protocol command would not). */
export type DaemonStatus = "current" | "stale" | "unreachable" | "orphan-file";
export interface DiscoveredDaemonProcess { pid: number; socketPath: string; uptimeSeconds?: number;}
export interface DaemonInfo { socketPath: string; pid?: number; uptimeSeconds?: number; version?: string; protocolVersion?: number; schemaId?: string; buildId?: string; executablePath?: string; pidSource?: "listener" | "hello"; sessionCount?: number; status: DaemonStatus; isDefault: boolean; hasTrackedWorkers?: boolean;}
const STATUS_ORDER: Record<DaemonStatus, number> = { current: 0, stale: 1, unreachable: 2, "orphan-file": 3,};const SHUTDOWN_QUIET_PERIOD_MS = 1000;const SHUTDOWN_CONVERGENCE_TIMEOUT_MS = 10_000;
export function evaluateShutdownQuietPeriod(now: number, quietSince: number | undefined): "complete" | "waiting" { if (quietSince !== undefined && now - quietSince >= SHUTDOWN_QUIET_PERIOD_MS) { return "complete"; } return "waiting";}
// Linux comm names (and thus the process name ss reports) are capped at 15 chars.const MAX_COMM_LENGTH = 15;
function processNameMatches(name: string, appName: string): boolean { return name === appName || appName.slice(0, MAX_COMM_LENGTH) === name;}
/** Parse `ss -lxp` output into the prime-agent daemons listening on unix sockets. */export function parseSsListeners(stdout: string, appName: string): DiscoveredDaemonProcess[] { const daemons: DiscoveredDaemonProcess[] = []; for (const line of stdout.split("\n")) { const fields = line.trim().split(/\s+/); if (fields[1] !== "LISTEN") { continue; } const socketPath = fields[4]; if (!socketPath?.startsWith("/")) { continue; } const owner = line.match(/users:\(\("([^"]+)",pid=(\d+)/); if (!owner || !processNameMatches(owner[1]!, appName)) { continue; } daemons.push({ pid: Number.parseInt(owner[2]!, 10), socketPath: normalizeSocketPath(socketPath) }); } return daemons;}
/** Parse `lsof -nP -F pn -U -a -c <app>` output into listening unix socket owners (macOS fallback). */export function parseLsofListeners(stdout: string): DiscoveredDaemonProcess[] { const daemons: DiscoveredDaemonProcess[] = []; const seen = new Set<string>(); let pid: number | undefined; for (const line of stdout.split("\n")) { const field = line[0]; const value = line.slice(1); if (field === "p") { pid = Number.parseInt(value, 10); } else if (field === "n" && pid !== undefined && value.startsWith("/")) { const socketPath = normalizeSocketPath(value); const key = `${pid}:${socketPath}`; if (!seen.has(key)) { seen.add(key); daemons.push({ pid, socketPath }); } } } return daemons;}
export function parsePrimeAgentProcessIds(stdout: string, appName: string): number[] { const pids: number[] = []; for (const line of stdout.split("\n")) { const match = line.trim().match(/^(\d+)\s+(\S+)(?:\s+(.*))?$/); if (!match) { continue; } const command = basename(match[2]!); const argv0 = basename(match[3]?.trim().split(/\s+/, 1)[0] ?? ""); if (processNameMatches(command, appName) || processNameMatches(argv0, appName)) { pids.push(Number.parseInt(match[1]!, 10)); } } return pids;}
export function mergeDiscoveredDaemonProcesses( ...groups: readonly DiscoveredDaemonProcess[][]): DiscoveredDaemonProcess[] { const byIdentity = new Map<string, DiscoveredDaemonProcess>(); for (const group of groups) { for (const daemon of group) { byIdentity.set(`${daemon.pid}:${daemon.socketPath}`, daemon); } } return [...byIdentity.values()];}
/** Parse `ps -o pid=,etimes=` output into a pid → uptime-seconds map. */export function parsePsEtimes(stdout: string): Map<number, number> { const uptimes = new Map<number, number>(); for (const line of stdout.split("\n")) { const match = line.trim().match(/^(\d+)\s+(\d+)$/); if (match) { uptimes.set(Number.parseInt(match[1]!, 10), Number.parseInt(match[2]!, 10)); } } return uptimes;}
/** * Listening prime-agent daemons that belong to this state root. The OS sweep * behind it sees every daemon the user can observe, including daemons from an * isolated HOME or a different agent dir, so the result is filtered before any * caller can list, signal or reap it. */function scanListeningDaemons(): DiscoveredDaemonProcess[] { const belongsToStateRoot = createDaemonStateRootMatcher(); return scanAllListeningDaemons().filter((daemon) => belongsToStateRoot(daemon.socketPath));}
function scanAllListeningDaemons(): DiscoveredDaemonProcess[] { if (process.platform === "win32") { return []; } const ss = spawnSyncHidden("ss", ["-lxp"], { encoding: "utf8" }); if (!ss.error && ss.status === 0 && typeof ss.stdout === "string") { return enrichUptimes(parseSsListeners(ss.stdout, APP_NAME)); } const lsof = spawnSyncHidden("lsof", ["-nP", "-F", "pn", "-U", "-a", "-c", APP_NAME], { encoding: "utf8" }); const byName = !lsof.error && typeof lsof.stdout === "string" ? parseLsofListeners(lsof.stdout) : []; let byPid: DiscoveredDaemonProcess[] = []; const ps = spawnSyncHidden("ps", ["-axo", "pid=,comm=,args="], { encoding: "utf8" }); if (!ps.error && ps.status === 0 && typeof ps.stdout === "string") { const pids = parsePrimeAgentProcessIds(ps.stdout, APP_NAME); if (pids.length > 0) { const lsofByPid = spawnSyncHidden("lsof", ["-nP", "-F", "pn", "-U", "-a", "-p", pids.join(",")], { encoding: "utf8", }); if (!lsofByPid.error && typeof lsofByPid.stdout === "string") { byPid = parseLsofListeners(lsofByPid.stdout); } } } return enrichUptimes(mergeDiscoveredDaemonProcesses(byName, byPid));}
function isDaemonProcessListening(pid: number, socketPath: string): boolean { const target = normalizeSocketPath(socketPath); return scanListeningDaemons().some((daemon) => daemon.pid === pid && daemon.socketPath === target);}
function enrichUptimes(daemons: DiscoveredDaemonProcess[]): DiscoveredDaemonProcess[] { const pids = daemons.map((daemon) => daemon.pid); if (pids.length === 0) { return daemons; } const ps = spawnSyncHidden("ps", ["-o", "pid=,etimes=", "-p", pids.join(",")], { encoding: "utf8" }); if (ps.error || typeof ps.stdout !== "string") { return daemons; } const uptimes = parsePsEtimes(ps.stdout); return daemons.map((daemon) => ({ ...daemon, uptimeSeconds: uptimes.get(daemon.pid) }));}
/** Socket files in the default socket dir (may be live daemons or orphaned files). */function scanSocketDir(): string[] { if (process.platform === "win32") { return []; } const dir = defaultDaemonSocketDir(); if (!existsSync(dir)) { return []; } const sockets: string[] = []; for (const entry of readdirSync(dir)) { const socketPath = join(dir, entry); try { if (lstatSync(socketPath).isSocket()) { sockets.push(normalizeSocketPath(socketPath)); } } catch { // Entry vanished between readdir and lstat; ignore. } } return sockets;}
interface ProbeResult { version?: string; protocolVersion?: number; schemaId?: string; runtime?: DaemonRuntimeIdentity; sessionCount?: number; supervisorPid?: number; supervisorProcessStartId?: string; reachable: boolean;}
async function probeDaemon(socketPath: string): Promise<ProbeResult> { const client = new DaemonClient(socketPath); try { await client.connect(300); } catch { client.close(); return { reachable: false }; } try { let version: string | undefined; let protocolVersion: number | undefined; let schemaId: string | undefined; let runtime: DaemonRuntimeIdentity | undefined; let supervisorPid: number | undefined; let supervisorProcessStartId: string | undefined; let greeted = false; try { const hello = await client.waitForHello(1500); version = hello.appVersion; protocolVersion = hello.protocol.version; schemaId = hello.schemaId; runtime = hello.runtime; supervisorPid = hello.supervisorPid; supervisorProcessStartId = hello.supervisorProcessStartId; greeted = true; } catch { // Connected but no recognizable greeting: an old/foreign daemon. } let sessionCount: number | undefined; try { const response = await client.request({ type: "list" }, greeted ? 30000 : 1500); if (response.success) { const sessions = (response.data as { sessions?: unknown })?.sessions; if (Array.isArray(sessions)) { sessionCount = sessions.length; } } } catch { // Leave sessionCount undefined when the daemon will not answer list. } return { version, protocolVersion, schemaId, runtime, sessionCount, supervisorPid, supervisorProcessStartId, reachable: true, }; } finally { client.close(); }}
function classifyReachable(probe: ProbeResult): DaemonStatus { if ( probe.protocolVersion === DAEMON_PROTOCOL_VERSION && probe.schemaId === DAEMON_SCHEMA_ID && probe.version === VERSION ) { return "current"; } return "stale";}
export function verifyHelloSupervisorPid( pid: number | undefined, expectedProcessStartId: string | undefined,): number | undefined { if (!Number.isInteger(pid) || pid === undefined || pid <= 0) { return undefined; } try { process.kill(pid, 0); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "EPERM") { return undefined; } } if (expectedProcessStartId) { const observedStartId = getProcessStartId(pid); if (observedStartId !== expectedProcessStartId) { return undefined; } } return pid;}
/** Discover every daemon in this state root and probe each for version + session count. */export async function discoverDaemons(): Promise<DaemonInfo[]> { const processBySocket = new Map<string, DiscoveredDaemonProcess>(); for (const daemon of scanListeningDaemons()) { if (isWorkerSocketPath(daemon.socketPath)) { continue; } processBySocket.set(daemon.socketPath, daemon); }
const workerSockets = new Set( findAllTrackedWorkers().map((worker) => normalizeSocketPath(worker.descriptor.supervisorSocketPath)), ); const sockets = new Set<string>([ ...processBySocket.keys(), ...scanSocketDir().filter((socketPath) => !isWorkerSocketPath(socketPath)), ...workerSockets, ]); const defaultSocket = normalizeSocketPath(defaultDaemonSocketPath());
const infos = await Promise.all( [...sockets].map(async (socketPath): Promise<DaemonInfo> => { const proc = processBySocket.get(socketPath); const probe = await probeDaemon(socketPath); const pid = proc?.pid ?? verifyHelloSupervisorPid(probe.supervisorPid, probe.supervisorProcessStartId); const hasTrackedWorkers = workerSockets.has(socketPath); const status: DaemonStatus = probe.reachable ? classifyReachable(probe) : proc || hasTrackedWorkers ? "unreachable" : "orphan-file"; return { socketPath, pid, uptimeSeconds: proc?.uptimeSeconds, version: probe.version, protocolVersion: probe.protocolVersion, schemaId: probe.schemaId, buildId: probe.runtime?.buildId, executablePath: probe.runtime?.launcherPath ?? probe.runtime?.entrypointPath ?? probe.runtime?.executablePath, ...(pid !== undefined ? { pidSource: proc ? ("listener" as const) : ("hello" as const) } : {}), sessionCount: probe.sessionCount, status, isDefault: socketPath === defaultSocket, ...(hasTrackedWorkers ? { hasTrackedWorkers: true } : {}), }; }), );
return sortDaemons(infos);}
export function sortDaemons(infos: DaemonInfo[]): DaemonInfo[] { return [...infos].sort((left, right) => { if (left.isDefault !== right.isDefault) { return left.isDefault ? -1 : 1; } const statusDelta = STATUS_ORDER[left.status] - STATUS_ORDER[right.status]; return statusDelta || left.socketPath.localeCompare(right.socketPath); });}
export async function runPs(json: boolean): Promise<void> { const daemons = await discoverDaemons(); if (json) { console.log(JSON.stringify(daemons, null, 2)); return; } if (daemons.length === 0) { console.log("No background services found."); return; } console.log(formatDaemonListTable(daemons));}
export type ReapAction = | { kind: "remove-file"; daemon: DaemonInfo } | { kind: "kill"; daemon: DaemonInfo } | { kind: "shutdown"; daemon: DaemonInfo } | { kind: "skip"; daemon: DaemonInfo; reason: string };
/** * Decide what to do with each discovered daemon (pure, no side effects). Reap * targets only clearly-safe daemons: orphaned socket files (including a stale * default daemon.sock with no live process), and reachable idle daemons on * non-default sockets. A reachable default daemon, and any reachable daemon with * live sessions, are never touched. * * Unreachable (hung) daemons can't report a session count, so they are only * ever killed with `force`, and even then never via a pid that backs more than * one discovered daemon (e.g. macOS lsof reports every unix socket a process * holds, so killing a shared pid could take down a reachable daemon with live * sessions). runReap additionally re-probes a kill candidate immediately before * the SIGTERM and backs off to the session-aware paths if it has since become * reachable, so a daemon that recovered with live sessions is never killed. */export function planReap(daemons: readonly DaemonInfo[], force: boolean): ReapAction[] { const pidCounts = new Map<number, number>(); for (const daemon of daemons) { if (daemon.pid !== undefined) { pidCounts.set(daemon.pid, (pidCounts.get(daemon.pid) ?? 0) + 1); } }
return daemons.map((daemon): ReapAction => { // An orphan socket file has no owning process, so removing it is safe even // on the default path (a stale daemon.sock left by a crash). Decide this // before the default guard so a dead default socket still gets cleaned up. if (daemon.status === "orphan-file") { return { kind: "remove-file", daemon }; } if (daemon.isDefault) { return { kind: "skip", daemon, reason: "default background service" }; } if (daemon.status === "unreachable") { if (!force || daemon.pid === undefined) { return { kind: "skip", daemon, reason: 'unreachable; use "prime-agent shutdown --force" to stop it' }; } if ((pidCounts.get(daemon.pid) ?? 0) > 1) { return { kind: "skip", daemon, reason: `unreachable; pid ${daemon.pid} also backs another daemon, not killing`, }; } return { kind: "kill", daemon }; } if (daemon.sessionCount !== 0) { return { kind: "skip", daemon, reason: `has ${daemon.sessionCount ?? "unknown"} session(s)` }; } return { kind: "shutdown", daemon }; });}
export function planShutdownAll(daemons: readonly DaemonInfo[], force: boolean): ReapAction[] { return daemons.map((daemon): ReapAction => { if (daemon.status === "orphan-file") { return { kind: "remove-file", daemon }; } if (daemon.status === "unreachable") { if (daemon.pid === undefined) { return force || !daemon.hasTrackedWorkers ? { kind: "remove-file", daemon } : { kind: "skip", daemon, reason: "has unreachable workers; use --force to kill" }; } return force ? { kind: "kill", daemon } : { kind: "skip", daemon, reason: "unreachable; use --force to kill" }; } return { kind: "shutdown", daemon }; });}
const SHUTDOWN_ALL_ACTION_ORDER: Record<ReapAction["kind"], number> = { shutdown: 0, "remove-file": 1, kill: 2, skip: 3,};
export type ShutdownConfirmationPlan = "none" | "prompt" | "json-error" | "tty-error";
export function planShutdownConfirmation( daemonCount: number, json: boolean, force: boolean, stdinIsTTY: boolean | undefined,): ShutdownConfirmationPlan { if (daemonCount === 0 || force) return "none"; if (json) return "json-error"; return stdinIsTTY ? "prompt" : "tty-error";}
export async function runShutdownAll(json: boolean, force: boolean): Promise<void> { const daemons = await discoverDaemons(); switch (planShutdownConfirmation(daemons.length, json, force, process.stdin.isTTY)) { case "json-error": process.exitCode = 1; console.log( JSON.stringify( { stopped: [], failed: daemons.map(({ socketPath }) => ({ socketPath, reason: 'confirmation required; use "prime-agent shutdown --force --json"', })), }, null, 2, ), ); return; case "tty-error": throw new Error( 'Shutdown requires confirmation in an interactive terminal. Use "prime-agent shutdown --force".', ); case "prompt": { const confirmed = await promptYesNo( "Stop every agent and background service? Active work will be interrupted.", ); if (!confirmed) { console.log(chalk.dim("Shutdown cancelled.")); return; } break; } case "none": break; } const admission = await acquireDaemonShutdownAdmission(); try { await runShutdownAllConverging(json, force, () => admission.assertOrRenew()); } finally { await admission.release(); }}
async function runShutdownAllConverging( json: boolean, force: boolean, assertAdmission: () => Promise<void>,): Promise<void> { const stopped: Array<{ socketPath: string; action: string }> = []; const failed: Array<{ socketPath: string; reason: string }> = []; const handledPids = new Set<number>(); const reportedFailures = new Set<string>();
if (force) { await stopHiddenSupervisors(stopped, failed, handledPids, reportedFailures, assertAdmission); } const daemons = (await discoverDaemons()).filter((daemon) => !isWorkerSocketPath(daemon.socketPath));
const actions = [...planShutdownAll(daemons, force)].sort( (left, right) => SHUTDOWN_ALL_ACTION_ORDER[left.kind] - SHUTDOWN_ALL_ACTION_ORDER[right.kind], );
for (const action of actions) { const { socketPath, pid } = action.daemon; if (pid !== undefined && handledPids.has(pid)) { await assertAdmission(); removeSocketFile(socketPath); stopped.push({ socketPath, action: `background service already stopped (pid ${pid})` }); if (force) { failed.push( ...(await forceStopTrackedWorkers(socketPath, assertAdmission)).map((reason) => ({ socketPath, reason, })), ); } continue; } switch (action.kind) { case "remove-file": { if ((await probeDaemon(socketPath)).reachable) { apply( await stopBackgroundService(socketPath, pid, handledPids, force, assertAdmission), socketPath, stopped, failed, ); } else { await assertAdmission(); if (removeSocketFile(socketPath)) { stopped.push({ socketPath, action: "removed stale socket file" }); } else { failed.push({ socketPath, reason: "could not remove socket file" }); } } break; } case "kill": { if ((await probeDaemon(socketPath)).reachable) { apply( await stopBackgroundService(socketPath, pid, handledPids, force, assertAdmission), socketPath, stopped, failed, ); } else if (isDaemonProcessListening(pid!, socketPath)) { await assertAdmission(); await forceKillDaemon(pid!); handledPids.add(pid!); await assertAdmission(); removeSocketFile(socketPath); stopped.push({ socketPath, action: `killed unreachable background service (pid ${pid})` }); } else { await assertAdmission(); removeSocketFile(socketPath); stopped.push({ socketPath, action: "background service already stopped" }); } break; } case "shutdown": apply( await stopBackgroundService(socketPath, pid, handledPids, force, assertAdmission), socketPath, stopped, failed, ); break; case "skip": failed.push({ socketPath, reason: action.reason }); break; } if (force && action.kind !== "skip") { failed.push( ...(await forceStopTrackedWorkers(socketPath, assertAdmission)).map((reason) => ({ socketPath, reason })), ); } }
if (force) { await terminateVerifiedResiduals(stopped, failed, handledPids, reportedFailures, assertAdmission); }
if (json) { if (failed.length > 0) { process.exitCode = 1; } console.log(JSON.stringify({ stopped, failed }, null, 2)); return; } if (stopped.length === 0 && failed.length === 0) { console.log("No background services found."); return; } for (const entry of stopped) { console.log(chalk.green(`stopped ${entry.socketPath}: ${entry.action}`)); } for (const entry of failed) { console.log(chalk.red(`failed ${entry.socketPath}: ${entry.reason}`)); } if (failed.length > 0) { process.exitCode = 1; }}
async function stopHiddenSupervisors( stopped: Array<{ socketPath: string; action: string }>, failed: Array<{ socketPath: string; reason: string }>, handledPids: Set<number>, reportedFailures: Set<string>, assertAdmission: () => Promise<void>,): Promise<void> { while (true) { // Renew before the scan: scanListeningDaemons blocks the event loop in synchronous ps/lsof/ss // calls, and the admission lease cannot refresh itself while that runs. await assertAdmission(); const listeners = scanListeningDaemons().filter((listener) => !isWorkerSocketPath(listener.socketPath)); const bySocket = new Map<string, DiscoveredDaemonProcess[]>(); for (const listener of listeners) { const group = bySocket.get(listener.socketPath) ?? []; group.push(listener); bySocket.set(listener.socketPath, group); } const hidden: DiscoveredDaemonProcess[] = []; for (const [socketPath, group] of bySocket) { if (new Set(group.map((listener) => listener.pid)).size < 2) { continue; } const currentPid = (await probeDaemon(socketPath)).supervisorPid; if (currentPid === undefined || !group.some((listener) => listener.pid === currentPid)) { recordShutdownFailure( failed, reportedFailures, socketPath, "could not identify the current same-path daemon", ); continue; } hidden.push(...group.filter((listener) => listener.pid !== currentPid)); } if (hidden.length === 0) { return; } const before = daemonListenerSignature(hidden); for (const listener of hidden) { if (await terminateVerifiedListener(listener, failed, reportedFailures, assertAdmission)) { handledPids.add(listener.pid); stopped.push({ socketPath: listener.socketPath, action: `stopped hidden daemon (pid ${listener.pid})` }); } } const afterHidden = scanListeningDaemons().filter( (listener) => !isWorkerSocketPath(listener.socketPath) && hidden.some((candidate) => candidate.pid === listener.pid && candidate.socketPath === listener.socketPath), ); if (afterHidden.length === 0 || daemonListenerSignature(afterHidden) === before) { return; } }}
async function terminateVerifiedResiduals( stopped: Array<{ socketPath: string; action: string }>, failed: Array<{ socketPath: string; reason: string }>, handledPids: Set<number>, reportedFailures: Set<string>, assertAdmission: () => Promise<void>,): Promise<void> { let previousSignature: string | undefined; let quietSince: number | undefined; const deadline = Date.now() + SHUTDOWN_CONVERGENCE_TIMEOUT_MS; while (true) { await assertAdmission(); const listeners = scanListeningDaemons(); const now = Date.now(); if (listeners.length === 0) { previousSignature = undefined; quietSince ??= now; const quietPeriod = evaluateShutdownQuietPeriod(now, quietSince); if (quietPeriod === "complete") { return; } await delay(100); continue; } quietSince = undefined; const signature = daemonListenerSignature(listeners); if (now >= deadline) { recordResidualListenerFailures(listeners, failed, reportedFailures, "kept respawning during shutdown"); return; } if (signature === previousSignature) { recordResidualListenerFailures(listeners, failed, reportedFailures, "remained after shutdown"); return; } previousSignature = signature; const seenPids = new Set<number>(); for (const listener of listeners) { if (seenPids.has(listener.pid)) { continue; } seenPids.add(listener.pid); const alreadyReported = handledPids.has(listener.pid); if (await terminateVerifiedListener(listener, failed, reportedFailures, assertAdmission)) { handledPids.add(listener.pid); if (!alreadyReported) { stopped.push({ socketPath: listener.socketPath, action: `stopped residual daemon process (pid ${listener.pid})`, }); } } } }}
function recordResidualListenerFailures( listeners: readonly DiscoveredDaemonProcess[], failed: Array<{ socketPath: string; reason: string }>, reportedFailures: Set<string>, reason: string,): void { for (const listener of listeners) { const processStartId = getProcessStartId(listener.pid); const identity = processStartId ? `pid ${listener.pid}, start ${processStartId}` : `pid ${listener.pid}, process identity unavailable`; recordShutdownFailure( failed, reportedFailures, listener.socketPath, `daemon ${reason} (${identity})${describeDaemonParent(listener.pid)}`, ); }}function describeDaemonParent(pid: number): string { const result = spawnSyncHidden("ps", ["-o", "ppid=,tty=,command=", "-p", String(pid)], { encoding: "utf8" }); if (result.error || result.status !== 0 || typeof result.stdout !== "string") { return ""; } const match = result.stdout.trim().match(/^(\d+)\s+(\S+)\s+(.+)$/); if (!match) { return ""; } return `; close parent PID ${match[1]} on ${match[2]} (${match[3]}) and retry shutdown`;}
async function terminateVerifiedListener( listener: DiscoveredDaemonProcess, failed: Array<{ socketPath: string; reason: string }>, reportedFailures: Set<string>, assertAdmission: () => Promise<void>,): Promise<boolean> { const processStartId = getProcessStartId(listener.pid); if (!processStartId) { recordShutdownFailure( failed, reportedFailures, listener.socketPath, `could not verify daemon process identity (pid ${listener.pid})`, ); return false; } if (getProcessStartId(listener.pid) !== processStartId) { return false; } await assertAdmission(); if (getProcessStartId(listener.pid) !== processStartId) { return false; } killDaemon(listener.pid); const deadline = Date.now() + 1000; while (getProcessStartId(listener.pid) === processStartId && Date.now() < deadline) { await delay(50); } if (getProcessStartId(listener.pid) === processStartId) { await assertAdmission(); if (getProcessStartId(listener.pid) !== processStartId) { return false; } try { process.kill(listener.pid, "SIGKILL"); } catch { // The verified process exited between the identity check and signal. } } return getProcessStartId(listener.pid) !== processStartId;}
function daemonListenerSignature(listeners: readonly DiscoveredDaemonProcess[]): string { return listeners .map((listener) => `${listener.pid}:${getProcessStartId(listener.pid) ?? "unknown"}:${listener.socketPath}`) .sort() .join("\n");}
function recordShutdownFailure( failed: Array<{ socketPath: string; reason: string }>, reportedFailures: Set<string>, socketPath: string, reason: string,): void { const key = `${socketPath}\0${reason}`; if (reportedFailures.has(key)) { return; } reportedFailures.add(key); failed.push({ socketPath, reason });}
export function isWorkerSocketPath(socketPath: string): boolean { return ( process.platform !== "win32" && resolve(dirname(socketPath)) === resolve(defaultDaemonSocketDir()) && basename(socketPath).startsWith("worker-") && basename(socketPath).endsWith(".sock") );}
async function stopBackgroundService( socketPath: string, pid: number | undefined, handledPids: Set<number>, force: boolean, assertAdmission: () => Promise<void>,): Promise<ReapOutcome> { await assertAdmission(); if (await shutdownDaemon(socketPath, force)) { if (pid !== undefined) { handledPids.add(pid); } return { reaped: `stopped background service${pid ? ` (pid ${pid})` : ""}` }; } if (!(await canConnectToSocket(socketPath, 250))) { await assertAdmission(); removeSocketFile(socketPath); return { reaped: "background service already stopped" }; } if (pid === undefined) { return { skipped: "still listening but no pid to kill" }; } if (!force) { return { skipped: "did not stop gracefully; retry with --force" }; } await assertAdmission(); await forceKillDaemon(pid); handledPids.add(pid); await assertAdmission(); removeSocketFile(socketPath); return { reaped: `force-killed unresponsive background service (pid ${pid})` };}
interface TrackedWorker { descriptor: DaemonWorkerDescriptor; descriptorPath: string;}
async function forceStopTrackedWorkers( supervisorSocketPath: string, assertAdmission: () => Promise<void>,): Promise<string[]> { const failures: string[] = []; for (const worker of findTrackedWorkers(supervisorSocketPath)) { const { descriptor } = worker; let cleanupWorkerRecords = await stopTrackedProcess(descriptor.pid, descriptor.processStartId, assertAdmission); if (!cleanupWorkerRecords) { failures.push(`could not safely stop worker ${descriptor.workerId} (pid ${descriptor.pid})`); } if (descriptor.orphanProcessJournalPath) { let orphans: ReturnType<typeof readActiveOrphanProcesses> = []; try { orphans = readActiveOrphanProcesses(descriptor.orphanProcessJournalPath, descriptor.pid); } catch (error) { failures.push(`could not read child process records for worker ${descriptor.workerId}: ${String(error)}`); } for (const orphan of orphans) { // Pid-only records go through the platform predicate (stopTrackedProcess needs a startId). if (orphan.processStartId === undefined) { if (shouldReapOrphanProcess(orphan)) { killOrphanProcess(orphan.pid); } continue; } if (!isOrphanProcessIdentityCurrent(orphan)) { continue; } if (process.platform === "win32") { // taskkill /T, like the sibling reapers: signalling only the shell pid leaves its descendants alive. await assertAdmission(); if (isOrphanProcessIdentityCurrent(orphan)) { killOrphanProcess(orphan.pid); if (isProcessAlive(orphan.pid)) { cleanupWorkerRecords = false; failures.push(`could not stop child process ${orphan.pid} for worker ${descriptor.workerId}`); } } continue; } if (!(await stopTrackedProcess(orphan.pid, orphan.processStartId, assertAdmission))) { cleanupWorkerRecords = false; failures.push(`could not stop child process ${orphan.pid} for worker ${descriptor.workerId}`); } } } if (cleanupWorkerRecords) { try { removeSocketFile(descriptor.socketPath); rmSync(worker.descriptorPath, { force: true }); rmSync(descriptor.recoveryJournalPath, { force: true }); if (descriptor.orphanProcessJournalPath) { rmSync(descriptor.orphanProcessJournalPath, { force: true }); } } catch (error) { failures.push(`could not clean up worker ${descriptor.workerId}: ${String(error)}`); } } } return failures;}
function findTrackedWorkers(supervisorSocketPath: string): TrackedWorker[] { return findAllTrackedWorkers().filter( (worker) => normalizeSocketPath(worker.descriptor.supervisorSocketPath) === normalizeSocketPath(supervisorSocketPath), );}
function findAllTrackedWorkers(agentDir: string = getAgentDir()): TrackedWorker[] { const root = join(agentDir, "daemon-workers"); if (!existsSync(root)) { return []; } const workers: TrackedWorker[] = []; let directoryNames: string[]; try { directoryNames = readdirSync(root); } catch { return []; } for (const directoryName of directoryNames) { const directory = join(root, directoryName); try { if (!lstatSync(directory).isDirectory()) { continue; } } catch { continue; } let fileNames: string[]; try { fileNames = readdirSync(directory); } catch { continue; } for (const fileName of fileNames) { if (!fileName.endsWith(".json")) { continue; } const descriptorPath = join(directory, fileName); try { const value: unknown = JSON.parse(readFileSync(descriptorPath, "utf8")); if (isTrackedWorkerDescriptor(value)) { workers.push({ descriptor: value, descriptorPath }); } } catch { // Invalid or concurrently removed descriptors are not safe shutdown targets. } } } return workers;}
function isTrackedWorkerDescriptor(value: unknown): value is DaemonWorkerDescriptor { if (!value || typeof value !== "object") { return false; } const descriptor = value as Partial<DaemonWorkerDescriptor>; return ( (descriptor.version === 1 || descriptor.version === 2) && typeof descriptor.supervisorSocketPath === "string" && typeof descriptor.workerId === "string" && Number.isInteger(descriptor.pid) && (descriptor.pid ?? 0) > 0 && (descriptor.processStartId === undefined || typeof descriptor.processStartId === "string") && typeof descriptor.socketPath === "string" && typeof descriptor.recoveryJournalPath === "string" );}
async function stopTrackedProcess( pid: number, expectedStartId: string | undefined, assertAdmission: () => Promise<void>,): Promise<boolean> { if (trackedProcessStopped(pid)) { return true; } if (!expectedStartId || !trackedLeaderIdentityCurrent(pid, expectedStartId)) { return false; } await assertAdmission(); if (!trackedLeaderIdentityCurrent(pid, expectedStartId)) { return false; } signalProcessGroupIfHeld(pid, "SIGTERM"); let deadline = Date.now() + 500; while (!trackedProcessStopped(pid) && Date.now() < deadline) { await delay(25); } if (trackedProcessStopped(pid)) { return true; } await assertAdmission(); if (!trackedLeaderIdentityCurrent(pid, expectedStartId)) { return false; } signalProcessGroupIfHeld(pid, "SIGKILL"); deadline = Date.now() + 1000; while (!trackedProcessStopped(pid) && Date.now() < deadline) { await delay(25); } return trackedProcessStopped(pid);}
/** A GROUP stop completes when the leader is gone AND no live member remains; unreaped zombies do not block it. */function trackedProcessStopped(pid: number): boolean { return !isProcessAlive(pid) && !processGroupHasLiveMember(pid);}
/** Identity gates guard pid reuse, so they apply only while the leader exists; a pgid cannot be reused while members hold it. */function trackedLeaderIdentityCurrent(pid: number, expectedStartId: string): boolean { if (!processIdExists(pid)) { return true; } return getProcessStartId(pid) === expectedStartId;}
export function reapStaleWorkerSessions( agentDir: string = getAgentDir(),): Array<{ socketPath: string; action: string }> { const reaped: Array<{ socketPath: string; action: string }> = []; const leaseRoot = join(agentDir, "session-leases");
for (const { descriptor, descriptorPath } of findAllTrackedWorkers(agentDir)) { const isDead = !isProcessAlive(descriptor.pid) || (descriptor.processStartId !== undefined && !trackedLeaderIdentityCurrent(descriptor.pid, descriptor.processStartId)); if (!isDead || (descriptor.lifecycle !== "recovering" && (descriptor.consecutiveFailures ?? 0) === 0)) continue; const sessionFile = descriptor.sessionFile ?? descriptor.createCommand?.sessionPath; const sessionName = sessionFile ? basename(sessionFile).replace(/\.jsonl$/, "") : (descriptor.rootSessionId ?? descriptor.rootActiveSessionId ?? "unknown");
rmSync(descriptorPath, { force: true }); if (descriptor.recoveryJournalPath) rmSync(descriptor.recoveryJournalPath, { force: true }); if (descriptor.orphanProcessJournalPath) rmSync(descriptor.orphanProcessJournalPath, { force: true });
if (sessionFile && existsSync(leaseRoot)) { const leaseKey = createHash("sha256").update(canonicalSessionPath(sessionFile)).digest("hex"); const leaseDir = join(leaseRoot, `${leaseKey}.lock`); rmSync(leaseDir, { recursive: true, force: true }); rmSync(`${leaseDir}.guard`, { force: true }); } if (descriptor.socketPath) rmSync(descriptor.socketPath, { force: true });
reaped.push({ socketPath: `session ${sessionName}`, action: "cleared stale recovering worker, lease, and socket", }); }
if (existsSync(leaseRoot)) { let entries: string[] = []; try { entries = readdirSync(leaseRoot); } catch {} for (const entry of entries) { if (!entry.endsWith(".lock")) continue; const dir = join(leaseRoot, entry); const ownerPath = join(dir, "owner.json"); if (!existsSync(ownerPath)) continue; try { const owner = JSON.parse(readFileSync(ownerPath, "utf8")) as { pid?: number; processStartId?: string; sessionPath?: string; }; if ( typeof owner.pid === "number" && (!isProcessAlive(owner.pid) || (owner.processStartId !== undefined && !trackedLeaderIdentityCurrent(owner.pid, owner.processStartId))) ) { rmSync(dir, { recursive: true, force: true }); rmSync(`${dir}.guard`, { force: true }); const sName = owner.sessionPath ? basename(owner.sessionPath).replace(/\.jsonl$/, "") : entry; if (!reaped.some((r) => r.socketPath === `session ${sName}`)) { reaped.push({ socketPath: `session ${sName}`, action: "cleared stale session lease" }); } } } catch {} } }
return reaped;}
export async function runReap(json: boolean, force: boolean): Promise<void> { const daemons = await discoverDaemons(); const reaped: Array<{ socketPath: string; action: string }> = [...reapStaleWorkerSessions()]; const skipped: Array<{ socketPath: string; reason: string }> = [];
for (const action of planReap(daemons, force)) { const { socketPath, pid } = action.daemon; switch (action.kind) { case "skip": skipped.push({ socketPath, reason: action.reason }); break; case "remove-file": { // Re-probe before unlinking: a path marked orphan-file at discovery // may have since become a live listener. Only remove it if it is // still unreachable, so we never delete a socket a daemon is using. if ((await probeDaemon(socketPath)).reachable) { skipped.push({ socketPath, reason: "now reachable; not removing socket file" }); } else if (removeSocketFile(socketPath)) { reaped.push({ socketPath, action: "removed stale socket file" }); } else { skipped.push({ socketPath, reason: "could not remove socket file" }); } break; } case "kill": { // Re-probe right before killing: discovery and this kill happen at // different moments, so a daemon classified "unreachable" may have // since started answering. Never SIGTERM one that now responds; // defer to the session-aware shutdown path instead. const recheck = await probeDaemon(socketPath); if (!recheck.reachable) { killDaemon(pid!); removeSocketFile(socketPath); reaped.push({ socketPath, action: `killed unreachable daemon (pid ${pid})` }); } else { apply(await reapReachableDaemon(socketPath, pid), socketPath, reaped, skipped); } break; } case "shutdown": apply(await reapReachableDaemon(socketPath, pid), socketPath, reaped, skipped); break; } }
if (json) { console.log(JSON.stringify({ reaped, skipped }, null, 2)); return; } if (reaped.length === 0 && skipped.length === 0) { console.log("No background services found."); return; } for (const entry of reaped) { console.log(chalk.green(`reaped ${entry.socketPath}: ${entry.action}`)); } for (const entry of skipped) { console.log(chalk.dim(`kept ${entry.socketPath}: ${entry.reason}`)); }}
type ReapOutcome = { reaped: string } | { skipped: string };
function apply( outcome: ReapOutcome, socketPath: string, reaped: Array<{ socketPath: string; action: string }>, skipped: Array<{ socketPath: string; reason: string }>,): void { if ("reaped" in outcome) { reaped.push({ socketPath, action: outcome.reaped }); } else { skipped.push({ socketPath, reason: outcome.skipped }); }}
/** * Gracefully stop a daemon, but only after a fresh probe confirms it is idle. * Discovery and reap happen at different moments, so the session count is * re-checked here to avoid stopping a daemon that gained a session in between. */async function reapReachableDaemon(socketPath: string, pid: number | undefined): Promise<ReapOutcome> { const probe = await probeDaemon(socketPath); if (!probe.reachable) { return { skipped: "no longer reachable" }; } if (probe.sessionCount !== 0) { return { skipped: `now has ${probe.sessionCount ?? "unknown"} session(s)` }; } return (await shutdownDaemon(socketPath, false)) ? { reaped: `stopped idle background service${pid ? ` (pid ${pid})` : ""}` } : { skipped: "shutdown request failed" };}
function removeSocketFile(socketPath: string): boolean { try { if (existsSync(socketPath)) { unlinkSync(socketPath); } return true; } catch { return false; }}
function killDaemon(pid: number): void { try { process.kill(pid, "SIGTERM"); } catch { // Process already gone or not permitted; the socket file cleanup still runs. }}
async function forceKillDaemon(pid: number): Promise<void> { killDaemon(pid); const deadline = Date.now() + 1000; while (Date.now() < deadline) { if (!isProcessAlive(pid)) { return; } await delay(50); } try { process.kill(pid, "SIGKILL"); } catch { // Process already exited between the liveness check and the kill. }}
function delay(ms: number): Promise<void> { return new Promise((resolve) => setTimeout(resolve, ms));}
async function canConnectToSocket(socketPath: string, timeoutMs: number): Promise<boolean> { const client = new DaemonClient(socketPath); try { await client.connect(timeoutMs); return true; } catch { return false; } finally { client.close(); }}
/** * Ask a daemon to shut down and confirm it actually stopped listening. The * shutdown ack alone is not proof, so success is reported only once the socket * stops accepting connections. */async function shutdownDaemon(socketPath: string, force: boolean): Promise<boolean> { const client = new DaemonClient(socketPath); try { await client.connect(1000); } catch { client.close(); return false; } try { await client.request({ type: "shutdown", force }, 1500); } catch { // The daemon may still stop; the connectivity check below is the source of truth. } finally { client.close(); }
const deadline = Date.now() + 5000; while (Date.now() < deadline) { if (!(await canConnectToSocket(socketPath, 250))) { return true; } await delay(50); } return false;}