perf(mcp): let the refresh launcher watch its server instead of a watchdog process

__tests__/fixtures/refresh-server.cjs

@@ -12,6 +12,8 @@ let initialized = false;
 let initializeParams;
 let hostQuestion;
 process.stderr.write(`FIXTURE_PID=${process.pid}\n`);
+// Arms the real liveness watchdog, as `serve --mcp` does.
+if (options.liveness) require(options.liveness).installMainThreadWatchdog();
 const lines = readline.createInterface({ input: process.stdin });
 lines.on("line", (line) => {
   const message = JSON.parse(line);
@@ -44,6 +46,7 @@ lines.on("line", (line) => {
   } else if (message.method === "tools/call") {
     const name = message.params.name;
     if (name === "crash") process.exit(9);
+    if (name === "wedge") for (;;) {}
     // Dies under the first call only, like a daemon restarting mid-request.
     if (name === "exit-once") {
       const marker = path.join(dist, "exit-once.marker");

__tests__/liveness-watchdog.test.ts

@@ -1,12 +1,14 @@
-import { describe, it, expect, beforeAll } from 'vitest';
+import { describe, it, expect, beforeAll, afterEach, vi } from 'vitest';
 import { spawn } from 'child_process';
 import * as fs from 'fs';
+import * as os from 'os';
 import * as path from 'path';
 import {
   parseWatchdogTimeoutMs,
   deriveCheckIntervalMs,
   installMainThreadWatchdog,
   DEFAULT_WATCHDOG_TIMEOUT_MS,
+  LivenessMonitor,
 } from '../src/mcp/liveness-watchdog';
 
 describe('config parsing', () => {
@@ -211,3 +213,51 @@ describe('liveness watchdog (spawned, real watchdog process)', () => {
     expect(code).toBe(3);
   }, 12000);
 });
+
+describe('LivenessMonitor (the refresh launcher side)', () => {
+  afterEach(() => vi.useRealTimers());
+
+  function armed(progressPaths: string[] = []) {
+    vi.useFakeTimers();
+    const kills: string[] = [];
+    const monitor = new LivenessMonitor((notice) => kills.push(notice));
+    monitor.receive(JSON.stringify({ timeoutMs: 1000, capMs: 3000, progressPaths }));
+    return { monitor, kills };
+  }
+
+  it('kills after a silent timeout and not while heartbeats arrive', () => {
+    const { monitor, kills } = armed();
+    for (let i = 0; i < 5; i++) { vi.advanceTimersByTime(900); monitor.receive(''); }
+    expect(kills).toEqual([]);
+    vi.advanceTimersByTime(1000);
+    expect(kills).toHaveLength(1);
+    expect(kills[0]).toContain('Main thread unresponsive for ~1s');
+  });
+
+  it('does not kill once disarmed', () => {
+    const { monitor, kills } = armed();
+    monitor.receive('disarm');
+    vi.advanceTimersByTime(5000);
+    expect(kills).toEqual([]);
+  });
+
+  it('defers while the watched file advances, up to the hard cap', () => {
+    const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'cg-monitor-'));
+    const db = path.join(dir, 'db');
+    fs.writeFileSync(db, 'a');
+    try {
+      const { kills } = armed([db]);
+      let size = 1;
+      const grow = () => fs.writeFileSync(db, 'a'.repeat(++size));
+      grow(); vi.advanceTimersByTime(1000);
+      expect(kills).toEqual([]);
+      grow(); vi.advanceTimersByTime(1000);
+      expect(kills).toEqual([]);
+      grow(); vi.advanceTimersByTime(1000);
+      expect(kills).toHaveLength(1);
+      expect(kills[0]).toContain('hard cap 3s reached');
+    } finally {
+      fs.rmSync(dir, { recursive: true, force: true });
+    }
+  });
+});

__tests__/refresh-launcher.test.ts

@@ -414,6 +414,29 @@ describe("isolated MCP refresh launcher", () => {
     expect(server.messages.filter((message) => message.id === 1)).toHaveLength(1);
   });
 
+  it.skipIf(process.platform === "win32")(
+    "kills a wedged child itself, without a watchdog process, and recovers on a fresh child",
+    async () => {
+      const server = await start({
+        options: { liveness: resolve(__dirname, "../dist/mcp/liveness-watchdog.js") },
+        env: { CODEGRAPH_WATCHDOG_TIMEOUT_MS: "600" },
+      });
+      const first = await server.call(1);
+      const pid = first.result.structuredContent.pid;
+      // The heartbeat goes to the launcher, so the server starts no child.
+      if (process.platform === "linux") {
+        expect(readFileSync(`/proc/${pid}/task/${pid}/children`, "utf8").trim()).toBe("");
+      }
+      // Both the child and its replay wedge, so the call is reported.
+      const wedged = await server.call(2, "wedge");
+      expect(wedged.error?.message).toContain("not replayed again");
+      expect(server.stderr()).toContain("Main thread unresponsive");
+      const recovered = await server.call(3);
+      expect(recovered.result.structuredContent.pid).not.toBe(pid);
+    },
+    20000,
+  );
+
   it("closes both children and queued requests when the host disconnects during preflight", async () => {
     const server = await start();
     server.deploy(B, { hang: true });

src/mcp/liveness-watchdog.ts

@@ -181,6 +181,165 @@ export interface WatchdogOptions {
   progressPaths?: string[];
 }
 
+/**
+ * The refresh launcher sets this to its own pid and opens a pipe on
+ * {@link LAUNCHER_LIVENESS_FD} for each server it starts. That server sends its
+ * heartbeat there instead of starting a watchdog process, and the launcher (a
+ * separate process, so the wedge cannot reach it either) kills it on silence and
+ * starts a fresh one. A process with any other parent that inherits the variable
+ * (a daemon, a CLI command) starts its own watchdog process as before.
+ */
+export const LAUNCHER_LIVENESS_ENV = 'CODEGRAPH_LAUNCHER_LIVENESS';
+export const LAUNCHER_LIVENESS_FD = 3;
+
+/**
+ * Lines on the launcher pipe: this object as JSON arms the watch, an empty line
+ * is a heartbeat, and `disarm` ends the watch.
+ */
+interface LivenessArm {
+  timeoutMs: number;
+  capMs: number;
+  progressPaths: string[];
+}
+
+/** Size + mtime of each watched file; a change is forward disk progress. */
+function progressSnapshot(paths: string[]): string {
+  let s = '';
+  for (const p of paths) {
+    try { const st = fs.statSync(p); s += st.size + ':' + st.mtimeMs + ';'; } catch { s += 'x;'; }
+  }
+  return s;
+}
+
+/**
+ * The launcher's side of the watch: the same timeout, disk-progress deferral and
+ * hard cap as the watchdog process ({@link CHILD_SOURCE}), driven by lines from
+ * the server's pipe. `kill` gets the notice to log and must end the server.
+ */
+export class LivenessMonitor {
+  private armed: LivenessArm | null = null;
+  private timer: NodeJS.Timeout | null = null;
+  private lastSnap = '';
+  private lastSnapAt = 0;
+  private silentSince: number | null = null;
+
+  constructor(private readonly kill: (notice: string) => void) {}
+
+  receive(line: string): void {
+    if (line === '') return this.beat();
+    if (line === 'disarm') return this.disarm();
+    let arm: Partial<LivenessArm>;
+    try { arm = JSON.parse(line) as Partial<LivenessArm>; } catch { return; }
+    if (!(Number(arm.timeoutMs) > 0) || !(Number(arm.capMs) > 0)) return;
+    const progressPaths = Array.isArray(arm.progressPaths) ? arm.progressPaths.filter((p) => typeof p === 'string') : [];
+    this.armed = { timeoutMs: Math.min(Number(arm.timeoutMs), MAX_TIMER_DELAY_MS), capMs: Number(arm.capMs), progressPaths };
+    this.lastSnap = progressSnapshot(progressPaths);
+    this.lastSnapAt = Date.now();
+    this.silentSince = null;
+    this.schedule();
+  }
+
+  disarm(): void {
+    this.armed = null;
+    if (this.timer) clearTimeout(this.timer);
+    this.timer = null;
+  }
+
+  private beat(): void {
+    if (!this.armed) return;
+    this.silentSince = null;
+    // Keep the baseline fresh while healthy (throttled: a stat per second).
+    if (this.armed.progressPaths.length && Date.now() - this.lastSnapAt >= 1000) {
+      this.lastSnap = progressSnapshot(this.armed.progressPaths);
+      this.lastSnapAt = Date.now();
+    }
+    this.schedule();
+  }
+
+  private schedule(): void {
+    if (this.timer) clearTimeout(this.timer);
+    this.timer = setTimeout(() => this.onTimeout(), this.armed!.timeoutMs);
+    this.timer.unref?.();
+  }
+
+  private onTimeout(): void {
+    const armed = this.armed;
+    if (!armed) return;
+    const notice = (extra: string): string =>
+      `Main thread unresponsive for ~${Math.round(armed.timeoutMs / 1000)}s${extra}; killing the wedged process so a fresh one can start (#850). Disable with CODEGRAPH_NO_WATCHDOG=1.`;
+    if (!armed.progressPaths.length) {
+      this.disarm();
+      return this.kill(notice(''));
+    }
+    const now = Date.now();
+    if (this.silentSince === null) this.silentSince = now - armed.timeoutMs;
+    const cur = progressSnapshot(armed.progressPaths);
+    if (cur !== this.lastSnap && now - this.silentSince < armed.capMs) {
+      // Blocked, but the DB files advance: a long store on slow storage.
+      this.lastSnap = cur;
+      this.schedule();
+      return;
+    }
+    this.disarm();
+    this.kill(notice(cur !== this.lastSnap ? ` despite ongoing disk activity (hard cap ${Math.round(armed.capMs / 1000)}s reached)` : ''));
+  }
+}
+
+/**
+ * Send heartbeats to the refresh launcher that started this process, or return
+ * null when there is none. Writes go through `fs.write` on the descriptor: a
+ * `net.Socket` opened on it delivers nothing under Bun 1.4.
+ */
+function launcherWatchdog(arm: LivenessArm, checkMs: number): WatchdogHandle | null {
+  if (process.platform === 'win32' || process.env[LAUNCHER_LIVENESS_ENV] !== String(process.ppid)) return null;
+  try {
+    const st = fs.fstatSync(LAUNCHER_LIVENESS_FD);
+    if (!st.isSocket() && !st.isFIFO()) return null;
+  } catch {
+    return null;
+  }
+  // One write at a time keeps the lines in order. A heartbeat is dropped while
+  // another line is still being written; that line already shows the loop turns.
+  const queue: string[] = [];
+  let writing = false;
+  let broken = false;
+  const send = (line: string, droppable = false): void => {
+    if (broken || (droppable && (writing || queue.length))) return;
+    queue.push(line + '\n');
+    if (!writing) flush();
+  };
+  const flush = (): void => {
+    const next = queue.shift();
+    if (next === undefined) { writing = false; return; }
+    writing = true;
+    fs.write(LAUNCHER_LIVENESS_FD, next, (err) => {
+      // The launcher can no longer watch this process (or is gone). Exit so a
+      // launcher that is still running replaces it, rather than leave a wedge unwatched.
+      if (err && err.code !== 'EAGAIN') {
+        broken = true;
+        queue.length = 0;
+        writing = false;
+        process.kill(process.pid, 'SIGKILL');
+        return;
+      }
+      flush();
+    });
+  };
+  send(JSON.stringify(arm));
+  const heartbeat = setInterval(() => send('', true), checkMs);
+  heartbeat.unref();
+  debug(`armed through launcher ${process.ppid}: timeoutMs=${arm.timeoutMs} checkMs=${checkMs} progressPaths=${arm.progressPaths.length}`);
+  let stopped = false;
+  return {
+    stop(): void {
+      if (stopped) return;
+      stopped = true;
+      clearInterval(heartbeat);
+      send('disarm');
+    },
+  };
+}
+
 /**
  * Install the main-thread liveness watchdog for a long-lived process. Returns a
  * handle to stop it, or `null` when disabled or when the child can't be spawned
@@ -195,6 +354,9 @@ export function installMainThreadWatchdog(options: WatchdogOptions = {}): Watchd
   const capMs = timeoutMs * PROGRESS_CAP_MULTIPLIER;
   const progressPaths = options.progressPaths ?? [];
 
+  const viaLauncher = launcherWatchdog({ timeoutMs, capMs, progressPaths }, checkMs);
+  if (viaLauncher) return viaLauncher;
+
   let child: ChildProcess;
   try {
     // No execArgv inheritance (unlike Worker), so the child carries none of our

src/mcp/refresh-launcher.ts

@@ -14,6 +14,8 @@ import { parseHostPpid, parsePpidPollMs, supervisionLostReason } from "./ppid-wa
 import { HOST_PPID_ENV } from "../extraction/node-runtime-flags";
 import { WRITER_LOCK_DEFER_ENV } from "./writer-lock";
 import { armStartupHandshakeTimeout } from "./startup-handshake";
+import { LAUNCHER_LIVENESS_ENV, LAUNCHER_LIVENESS_FD, LivenessMonitor } from "./liveness-watchdog";
+import type { Readable } from "stream";
 
 type Id = string | number | null;
 type Message = {
@@ -68,10 +70,12 @@ class Backend {
         ...args,
       ],
       {
-        stdio: ["pipe", "pipe", "pipe"],
+        // fd 3 carries the child's liveness heartbeat (see LAUNCHER_LIVENESS_ENV).
+        stdio: ["pipe", "pipe", "pipe", "pipe"],
         windowsHide: true,
         env: {
           ...process.env,
+          ...(process.platform === "win32" ? {} : { [LAUNCHER_LIVENESS_ENV]: String(process.pid) }),
           [HOST_PPID_ENV]: String(parseHostPpid(process.env[HOST_PPID_ENV]) ?? EARLY_PPID),
           // This child may start while the one it replaces still serves, and
           // holds the writer lock. It succeeds that child rather than
@@ -84,6 +88,22 @@ class Backend {
     this.child.once("error", (error) => this.fail(error));
     this.child.stdin.on("error", (error) => this.fail(error));
     this.child.once("close", () => this.fail(new Error("CodeGraph child exited")));
+    // Kill a child whose main thread stops turning; the close above then
+    // replays its calls on a fresh child.
+    const liveness = this.child.stdio[LAUNCHER_LIVENESS_FD] as Readable | null | undefined;
+    if (liveness) {
+      const monitor = new LivenessMonitor((notice) => {
+        process.stderr.write(`[${new Date().toISOString()}] [CodeGraph refresh] Child ${this.child.pid}: ${notice}\n`);
+        this.child.kill("SIGKILL");
+      });
+      // A broken channel leaves the child unwatched; replace it.
+      liveness.on("error", () => {
+        monitor.disarm();
+        this.child.kill("SIGKILL");
+      });
+      createInterface({ input: liveness }).on("line", (line) => monitor.receive(line));
+      this.child.once("close", () => monitor.disarm());
+    }
     const lines = createInterface({ input: this.child.stdout });
     lines.on("line", (line) => {
       const message = parse(line);