{"version":3,"file":"lifeguard.d.ts","sourceRoot":"","sources":["../../src/core/lifeguard.ts"],"names":[],"mappings":"AAAA,OAAO,KAAK,EAAE,YAAY,EAAE,MAAM,oBAAoB,CAAC;AACvD,OAAO,EAAE,YAAY,EAAE,MAAM,aAAa,CAAC;AAsC3C,MAAM,WAAW,gBAAgB;IAChC,GAAG,EAAE,MAAM,CAAC;IACZ,OAAO,EAAE,MAAM,CAAC;IAChB,UAAU,EAAE,MAAM,CAAC;IACnB,OAAO,EAAE,YAAY,CAAC;CACtB;AAED;;;;GAIG;AACH,qBAAa,iBAAkB,SAAQ,YAAY;IAClD,OAAO,CAAC,SAAS,CAAuC;IACxD,OAAO,CAAC,aAAa,CAA6B;IAClD,OAAO,CAAC,QAAQ,CAAqC;IACrD,4EAA4E;IAC5E,OAAO,CAAC,SAAS,CAA6B;IAC9C,gFAAgF;IAChF,OAAO,CAAC,aAAa,CAA6B;IAClD,OAAO,CAAC,aAAa,CAA+B;IACpD,+EAA+E;IAC/E,OAAO,CAAC,WAAW,CAAc;IACjC;;;;;OAKG;IACH,OAAO,CAAC,YAAY,CAAK;IACzB;;;;OAIG;IACH,OAAO,CAAC,OAAO,CAAqB;IACpC,OAAO,CAAC,QAAQ,CAAS;IACzB,OAAO,CAAC,QAAQ,CAAC,GAAG,CAAS;IAC7B,OAAO,CAAC,qBAAqB,CAAC,CAAa;IAE3C,YAAY,GAAG,EAAE,MAAM,EAOtB;IAED;;;;;OAKG;IACH,eAAe,CAAC,KAAK,EAAE,MAAM,GAAG,IAAI,CAEnC;IAED;;;;;OAKG;IACH,OAAO,CAAC,cAAc;IAMtB;;;OAGG;IACH,OAAO,CAAC,OAAO,EAAE,MAAM,EAAE,UAAU,EAAE,MAAM,EAAE,IAAI,EAAE,YAAY,GAAG,IAAI,CAwBrE;IAED,+CAA+C;IAC/C,eAAe,CAAC,OAAO,EAAE,MAAM,GAAG,IAAI,CAIrC;IAED,0DAA0D;IAC1D,eAAe,CAAC,OAAO,EAAE,MAAM,GAAG,MAAM,GAAG,IAAI,CAE9C;IAED,qDAAqD;IACrD,YAAY,CAAC,OAAO,EAAE,MAAM,GAAG,OAAO,CAErC;IAED,iDAAiD;IACjD,OAAO,IAAI,IAAI,CA4Bd;IAED,OAAO,CAAC,eAAe;IAqBvB,OAAO,CAAC,aAAa;IAsBrB,OAAO,CAAC,aAAa;IAiCrB;;;;;;;OAOG;IACH,OAAO,CAAC,QAAQ;IAQhB,OAAO,CAAC,OAAO;IAaf,OAAO,CAAC,uBAAuB;IAQ/B,OAAO,CAAC,gBAAgB;IAuBxB,OAAO,CAAC,cAAc;IAyBtB,OAAO,CAAC,aAAa;IAcrB,OAAO,CAAC,IAAI;CAoBZ","sourcesContent":["import type { ChildProcess } from \"node:child_process\";\nimport { EventEmitter } from \"node:events\";\nimport { existsSync, readdirSync, readFileSync, rmdirSync, statSync, unlinkSync } from \"node:fs\";\nimport { join } from \"node:path\";\nimport { getDispatchRoot } from \"../config.js\";\nimport { killProcessTree } from \"../utils/shell.js\";\nimport { agentLog } from \"./agent-log.js\";\n\nconst TIMEOUTS_MS: Record<string, number> = {\n\texplore: 5 * 60 * 1000,\n\tedit: 10 * 60 * 1000,\n\ttest: 10 * 60 * 1000,\n\treview: 8 * 60 * 1000,\n\tdoc: 5 * 60 * 1000,\n};\n\nconst HEARTBEAT_MISS_THRESHOLD_MS = 60000;\nconst HEARTBEAT_CHECK_INTERVAL_MS = 5000;\nconst PARENT_SHUTDOWN_GRACE_MS = 5000;\n\n/**\n * Concurrency tolerance. Each *additional* concurrently-monitored subagent adds\n * this fraction to the heartbeat-miss and hard-timeout budgets.\n *\n * When several subagents run at once (plus background MCP tools), they saturate\n * the CPU and starve the parent's event loop: its `setInterval` heartbeat check\n * fires late, and it cannot read a child's `{\"ping\":true}` line in time even\n * though the child is healthy and still working. Scaling the budgets by load\n * stops that contention from false-positive SIGKILLing healthy subagents — the\n * failure the demo hit when running many agents + MCP tools in the background.\n */\nconst LOAD_TOLERANCE_PER_PROCESS = 0.5;\n\n/**\n * Hard ceiling on the load tolerance multiplier, so a genuinely stuck subagent\n * is still eventually reaped no matter how busy the pool is.\n */\nconst MAX_LOAD_MULTIPLIER = 4;\n\nexport interface LifeguardProcess {\n\tpid: number;\n\ttask_id: string;\n\tagent_type: string;\n\tprocess: ChildProcess;\n}\n\n/**\n * Monitors running subagent processes for heartbeats, hard timeouts,\n * and parent-exit cleanup. Emits \"stalled\" and \"timeout\" events when\n * processes are terminated.\n */\nexport class SubagentLifeguard extends EventEmitter {\n\tprivate processes = new Map<string, LifeguardProcess>();\n\tprivate lastHeartbeat = new Map<string, number>();\n\tprivate timeouts = new Map<string, NodeJS.Timeout>();\n\t/** When each task started, used to compute the load-scaled hard timeout. */\n\tprivate startedAt = new Map<string, number>();\n\t/** Per-agent base hard timeout (before load scaling), captured at monitor(). */\n\tprivate baseTimeoutMs = new Map<string, number>();\n\tprivate checkInterval: NodeJS.Timeout | null = null;\n\t/** Wall-clock time the heartbeat check last ran, to measure event-loop lag. */\n\tprivate lastCheckAt = Date.now();\n\t/**\n\t * Count of external in-process tasks (e.g. background MCP tools) running in the\n\t * parent alongside the monitored subagents. These don't show up in `processes`\n\t * but still saturate the parent's CPU/event loop, so they're folded into the\n\t * load multiplier. Updated by the pool via setExternalLoad().\n\t */\n\tprivate externalLoad = 0;\n\t/**\n\t * Tasks already reaped (stalled/timeout kill sent) but whose `exit` has not\n\t * fired yet. The heartbeat check runs every 5s, so without this a stalled task\n\t * would re-emit \"stalled\" (and re-kill) on every tick until the process exits.\n\t */\n\tprivate reaping = new Set<string>();\n\tprivate disposed = false;\n\tprivate readonly cwd: string;\n\tprivate parentShutdownHandler?: () => void;\n\n\tconstructor(cwd: string) {\n\t\tsuper();\n\t\tthis.cwd = cwd;\n\t\tthis.setupParentExitHandlers();\n\t\tthis.sweepOldAgents();\n\t\tthis.lastCheckAt = Date.now();\n\t\tthis.checkInterval = setInterval(() => this.checkHeartbeats(), HEARTBEAT_CHECK_INTERVAL_MS);\n\t}\n\n\t/**\n\t * Set the count of external in-process tasks (background MCP tools) sharing the\n\t * parent's CPU/event loop. Folded into loadMultiplier() so concurrent MCP work\n\t * widens the heartbeat/timeout budgets just like extra monitored subagents do.\n\t * Negative values are clamped to 0.\n\t */\n\tsetExternalLoad(count: number): void {\n\t\tthis.externalLoad = Math.max(0, Math.floor(count));\n\t}\n\n\t/**\n\t * Tolerance multiplier for the current load. 1 concurrent task → 1x; each extra\n\t * concurrent task adds LOAD_TOLERANCE_PER_PROCESS, capped at MAX_LOAD_MULTIPLIER.\n\t * Concurrency counts both monitored subagents and external load (background MCP\n\t * tools), since both contend for the same parent event loop.\n\t */\n\tprivate loadMultiplier(): number {\n\t\tconst concurrent = this.processes.size + this.externalLoad;\n\t\tconst mult = 1 + Math.max(0, concurrent - 1) * LOAD_TOLERANCE_PER_PROCESS;\n\t\treturn Math.min(mult, MAX_LOAD_MULTIPLIER);\n\t}\n\n\t/**\n\t * Begin monitoring a child process. The process must emit a\n\t * `{\"ping\":true}` JSON line on stdout every 30 seconds.\n\t */\n\tmonitor(task_id: string, agent_type: string, proc: ChildProcess): void {\n\t\tif (this.disposed) return;\n\n\t\tconst pid = proc.pid ?? 0;\n\t\tthis.processes.set(task_id, { pid, task_id, agent_type, process: proc });\n\t\tthis.lastHeartbeat.set(task_id, Date.now());\n\n\t\tconst timeoutMs = TIMEOUTS_MS[agent_type] ?? TIMEOUTS_MS.explore;\n\t\tthis.startedAt.set(task_id, Date.now());\n\t\tthis.baseTimeoutMs.set(task_id, timeoutMs);\n\t\t// Arm the hard timeout scaled by current load. When it fires, handleTimeout\n\t\t// re-checks load and re-arms (up to MAX_LOAD_MULTIPLIER) if the pool is still\n\t\t// busy, so a slow-but-progressing subagent isn't killed for CPU contention.\n\t\tconst timeout = setTimeout(\n\t\t\t() => {\n\t\t\t\tthis.handleTimeout(task_id);\n\t\t\t},\n\t\t\tMath.round(timeoutMs * this.loadMultiplier()),\n\t\t);\n\t\tthis.timeouts.set(task_id, timeout);\n\n\t\tproc.once(\"exit\", () => {\n\t\t\tthis.untrack(task_id);\n\t\t});\n\t}\n\n\t/** Record a heartbeat for a monitored task. */\n\trecordHeartbeat(task_id: string): void {\n\t\tif (this.processes.has(task_id)) {\n\t\t\tthis.lastHeartbeat.set(task_id, Date.now());\n\t\t}\n\t}\n\n\t/** Get the last recorded heartbeat timestamp, or null. */\n\tlastHeartbeatAt(task_id: string): number | null {\n\t\treturn this.lastHeartbeat.get(task_id) ?? null;\n\t}\n\n\t/** True if the task is currently being monitored. */\n\tisMonitoring(task_id: string): boolean {\n\t\treturn this.processes.has(task_id);\n\t}\n\n\t/** Kill all monitored processes and clean up. */\n\tdispose(): void {\n\t\tif (this.disposed) return;\n\t\tthis.disposed = true;\n\n\t\tif (this.checkInterval) {\n\t\t\tclearInterval(this.checkInterval);\n\t\t\tthis.checkInterval = null;\n\t\t}\n\n\t\tfor (const timeout of this.timeouts.values()) {\n\t\t\tclearTimeout(timeout);\n\t\t}\n\t\tthis.timeouts.clear();\n\n\t\tfor (const monitored of this.processes.values()) {\n\t\t\tthis.killTree(monitored);\n\t\t}\n\t\tthis.processes.clear();\n\t\tthis.lastHeartbeat.clear();\n\t\tthis.startedAt.clear();\n\t\tthis.baseTimeoutMs.clear();\n\t\tthis.reaping.clear();\n\t\tthis.removeAllListeners();\n\n\t\tif (this.parentShutdownHandler) {\n\t\t\tprocess.removeListener(\"SIGINT\", this.parentShutdownHandler);\n\t\t\tprocess.removeListener(\"SIGTERM\", this.parentShutdownHandler);\n\t\t}\n\t}\n\n\tprivate checkHeartbeats(): void {\n\t\tconst now = Date.now();\n\t\t// Event-loop lag: how much later than scheduled this check actually ran.\n\t\t// Heavy CPU load (many concurrent subagents + background MCP tools) starves\n\t\t// the loop, delaying both this check *and* our reading of child heartbeats.\n\t\t// Forgive that gap so the parent's own starvation isn't charged against the\n\t\t// children as a missed heartbeat.\n\t\tconst loopLag = Math.max(0, now - this.lastCheckAt - HEARTBEAT_CHECK_INTERVAL_MS);\n\t\tthis.lastCheckAt = now;\n\n\t\tconst threshold = HEARTBEAT_MISS_THRESHOLD_MS * this.loadMultiplier() + loopLag;\n\t\tfor (const [task_id] of this.processes) {\n\t\t\tif (this.reaping.has(task_id)) continue;\n\t\t\tconst last = this.lastHeartbeat.get(task_id);\n\t\t\tif (last === undefined) continue;\n\t\t\tif (now - last > threshold) {\n\t\t\t\tthis.handleStalled(task_id);\n\t\t\t}\n\t\t}\n\t}\n\n\tprivate handleStalled(task_id: string): void {\n\t\tconst monitored = this.processes.get(task_id);\n\t\tif (!monitored || this.reaping.has(task_id)) return;\n\t\tthis.reaping.add(task_id);\n\n\t\t// Record why we reaped this child so a recurrence is diagnosable rather than\n\t\t// just \"stalled\": how long since the last heartbeat, and the load factors\n\t\t// (concurrent monitored subagents, threshold) that fed the decision.\n\t\tconst last = this.lastHeartbeat.get(task_id);\n\t\tconst silentMs = last === undefined ? -1 : Date.now() - last;\n\t\tagentLog(\n\t\t\t`[LIFEGUARD] stalled task_id=${task_id} agent=${monitored.agent_type} ` +\n\t\t\t\t`silent_ms=${silentMs} concurrent=${this.processes.size} ` +\n\t\t\t\t`load_mult=${this.loadMultiplier().toFixed(2)} base_threshold_ms=${HEARTBEAT_MISS_THRESHOLD_MS}`,\n\t\t);\n\n\t\tthis.killTree(monitored);\n\n\t\tthis.emit(\"stalled\", { task_id, pid: monitored.pid });\n\t\t// Process exit handler will call untrack()\n\t}\n\n\tprivate handleTimeout(task_id: string): void {\n\t\tconst monitored = this.processes.get(task_id);\n\t\tif (!monitored || this.reaping.has(task_id)) return;\n\n\t\t// Under load the wall-clock timer can fire while the subagent is still doing\n\t\t// real work — its turns are just slow because the CPU is shared. Re-arm rather\n\t\t// than kill, up to a hard ceiling (base * MAX_LOAD_MULTIPLIER) so a genuinely\n\t\t// stuck agent still terminates.\n\t\tconst started = this.startedAt.get(task_id) ?? Date.now();\n\t\tconst base = this.baseTimeoutMs.get(task_id) ?? TIMEOUTS_MS[monitored.agent_type] ?? TIMEOUTS_MS.explore;\n\t\tconst elapsed = Date.now() - started;\n\t\tconst ceiling = base * MAX_LOAD_MULTIPLIER;\n\t\tif (this.loadMultiplier() > 1 && elapsed < ceiling) {\n\t\t\tconst remaining = ceiling - elapsed;\n\t\t\tconst next = Math.min(\n\t\t\t\tMath.round(base * this.loadMultiplier()),\n\t\t\t\tMath.max(HEARTBEAT_CHECK_INTERVAL_MS, remaining),\n\t\t\t);\n\t\t\tthis.timeouts.set(\n\t\t\t\ttask_id,\n\t\t\t\tsetTimeout(() => this.handleTimeout(task_id), next),\n\t\t\t);\n\t\t\treturn;\n\t\t}\n\n\t\tthis.reaping.add(task_id);\n\t\tthis.killTree(monitored);\n\n\t\tthis.emit(\"timeout\", { task_id, pid: monitored.pid });\n\t\tthis.timeouts.delete(task_id);\n\t\t// Process exit handler will call untrack()\n\t}\n\n\t/**\n\t * Kill a monitored subagent and everything it spawned. The pool spawns\n\t * children detached on POSIX (each leads its own process group), so the\n\t * group/tree kill reaches nested grandchildren (the subagent's bash commands,\n\t * its own subagents) that a single-PID SIGKILL would orphan. A pid of 0 means\n\t * the spawn never produced a process — nothing to kill (and `kill(-0)` would\n\t * signal our own process group).\n\t */\n\tprivate killTree(monitored: LifeguardProcess): void {\n\t\tif (monitored.pid > 0) {\n\t\t\tkillProcessTree(monitored.pid);\n\t\t} else if (!monitored.process.killed) {\n\t\t\tmonitored.process.kill(\"SIGKILL\");\n\t\t}\n\t}\n\n\tprivate untrack(task_id: string): void {\n\t\tconst timeout = this.timeouts.get(task_id);\n\t\tif (timeout) {\n\t\t\tclearTimeout(timeout);\n\t\t\tthis.timeouts.delete(task_id);\n\t\t}\n\t\tthis.processes.delete(task_id);\n\t\tthis.lastHeartbeat.delete(task_id);\n\t\tthis.startedAt.delete(task_id);\n\t\tthis.baseTimeoutMs.delete(task_id);\n\t\tthis.reaping.delete(task_id);\n\t}\n\n\tprivate setupParentExitHandlers(): void {\n\t\tconst shutdown = () => this.gracefulShutdown();\n\t\tthis.parentShutdownHandler = shutdown;\n\t\tprocess.setMaxListeners(Math.max(process.getMaxListeners(), 20));\n\t\tprocess.once(\"SIGINT\", shutdown);\n\t\tprocess.once(\"SIGTERM\", shutdown);\n\t}\n\n\tprivate gracefulShutdown(): void {\n\t\t// SIGTERM each child's process group (POSIX; the pool spawns them detached\n\t\t// as group leaders) so their own descendants get the graceful signal too.\n\t\tfor (const monitored of this.processes.values()) {\n\t\t\ttry {\n\t\t\t\tif (process.platform !== \"win32\" && monitored.pid > 0) {\n\t\t\t\t\tprocess.kill(-monitored.pid, \"SIGTERM\");\n\t\t\t\t} else if (!monitored.process.killed) {\n\t\t\t\t\tmonitored.process.kill(\"SIGTERM\");\n\t\t\t\t}\n\t\t\t} catch {\n\t\t\t\t// Already gone.\n\t\t\t}\n\t\t}\n\n\t\t// Kill whole trees after the grace period\n\t\tsetTimeout(() => {\n\t\t\tfor (const monitored of this.processes.values()) {\n\t\t\t\tthis.killTree(monitored);\n\t\t\t}\n\t\t}, PARENT_SHUTDOWN_GRACE_MS).unref();\n\t}\n\n\tprivate sweepOldAgents(): void {\n\t\tconst dispatchDir = getDispatchRoot(this.cwd);\n\t\tif (!existsSync(dispatchDir)) return;\n\n\t\tconst now = Date.now();\n\t\tconst cutoff = 24 * 60 * 60 * 1000; // 24 hours\n\n\t\tfor (const entry of readdirSync(dispatchDir)) {\n\t\t\tconst entryPath = join(dispatchDir, entry);\n\t\t\ttry {\n\t\t\t\tconst stats = statSync(entryPath);\n\t\t\t\tif (!stats.isDirectory()) continue;\n\n\t\t\t\tif (now - stats.mtimeMs > cutoff) {\n\t\t\t\t\tconst hasRunningPid = this.hasRunningPid(entryPath);\n\t\t\t\t\tif (!hasRunningPid) {\n\t\t\t\t\t\tthis.rmrf(entryPath);\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t} catch {\n\t\t\t\t// Ignore errors for individual entries\n\t\t\t}\n\t\t}\n\t}\n\n\tprivate hasRunningPid(dir: string): boolean {\n\t\tconst pidFile = join(dir, \"pid\");\n\t\tif (!existsSync(pidFile)) return false;\n\n\t\ttry {\n\t\t\tconst pid = Number.parseInt(readFileSync(pidFile, \"utf-8\"), 10);\n\t\t\tif (Number.isNaN(pid)) return false;\n\t\t\tprocess.kill(pid, 0); // Check if process exists\n\t\t\treturn true;\n\t\t} catch {\n\t\t\treturn false;\n\t\t}\n\t}\n\n\tprivate rmrf(dir: string): void {\n\t\tfor (const entry of readdirSync(dir)) {\n\t\t\tconst entryPath = join(dir, entry);\n\t\t\ttry {\n\t\t\t\tconst stats = statSync(entryPath);\n\t\t\t\tif (stats.isDirectory()) {\n\t\t\t\t\tthis.rmrf(entryPath);\n\t\t\t\t} else {\n\t\t\t\t\tunlinkSync(entryPath);\n\t\t\t\t}\n\t\t\t} catch {\n\t\t\t\t// Ignore\n\t\t\t}\n\t\t}\n\t\ttry {\n\t\t\trmdirSync(dir);\n\t\t} catch {\n\t\t\t// Ignore\n\t\t}\n\t}\n}\n"]}