feat(FN-5189): complete Step 2 — implement process supervisor

Fusion-Task-Id: FN-5189
Fusion-Task-Lineage: 4caa3f0a-af81-4c60-88e8-de229ed72e08

packages/core/src/__tests__/fixtures/process-supervisor-child.mjs

@@ -0,0 +1,30 @@
+#!/usr/bin/env node
+import { spawn } from "node:child_process";
+import { writeFileSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+
+const mode = process.argv[2] ?? "keepalive";
+const pidFile = process.argv[3];
+const extraFile = process.argv[4];
+
+if (pidFile) {
+  writeFileSync(pidFile, String(process.pid), "utf8");
+}
+
+if (mode === "exit-immediately") {
+  process.exit(0);
+}
+
+if (mode === "ignore-term") {
+  process.on("SIGTERM", () => {});
+}
+
+if (mode === "spawn-child") {
+  const selfPath = fileURLToPath(import.meta.url);
+  const grandchild = spawn(process.execPath, [selfPath, "keepalive", extraFile].filter(Boolean), {
+    stdio: "ignore",
+  });
+  grandchild.unref();
+}
+
+setInterval(() => {}, 1_000);

packages/core/src/__tests__/process-supervisor.test.ts

@@ -0,0 +1,151 @@
+import { mkdtempSync, readFileSync, rmSync } from "node:fs";
+import os from "node:os";
+import { join } from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import {
+  __getProcessSupervisorStateForTests,
+  __resetProcessSupervisorForTests,
+  __terminateSupervisedChildrenForTests,
+  superviseSpawn,
+} from "../process-supervisor.js";
+
+const fixturePath = join(import.meta.dirname, "fixtures", "process-supervisor-child.mjs");
+
+function isAlive(pid: number): boolean {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise<void> {
+  const startedAt = Date.now();
+  while (true) {
+    try {
+      if (predicate()) {
+        return;
+      }
+    } catch {
+      // Retry until the timeout; many predicates wait on files to appear.
+    }
+    if (Date.now() - startedAt > timeoutMs) {
+      throw new Error("Timed out waiting for condition");
+    }
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  }
+}
+
+describe("process-supervisor", () => {
+  const tempDirs: string[] = [];
+
+  afterEach(async () => {
+    await __terminateSupervisedChildrenForTests("afterEach");
+    __resetProcessSupervisorForTests();
+    vi.restoreAllMocks();
+    for (const dir of tempDirs.splice(0)) {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  });
+
+  it("registers a child and deregisters it after natural exit", async () => {
+    const child = superviseSpawn(process.execPath, [fixturePath, "exit-immediately"], {
+      stdio: "ignore",
+      maxLifetimeMs: 1_000,
+    });
+
+    expect(__getProcessSupervisorStateForTests()).toEqual({ registrySize: 1, handlersInstalled: true });
+    await expect(child.waitExit()).resolves.toEqual({ code: 0, signal: null });
+    await waitFor(() => __getProcessSupervisorStateForTests().registrySize === 0);
+  });
+
+  it("cascades SIGTERM to the supervised process group", async () => {
+    if (process.platform === "win32") {
+      return;
+    }
+
+    const root = mkdtempSync(join(os.tmpdir(), "fn-process-supervisor-"));
+    tempDirs.push(root);
+    const parentPidFile = join(root, "parent.pid");
+    const grandchildPidFile = join(root, "grandchild.pid");
+
+    const child = superviseSpawn(process.execPath, [fixturePath, "spawn-child", parentPidFile, grandchildPidFile], {
+      stdio: "ignore",
+      killGraceMs: 100,
+      maxLifetimeMs: 5_000,
+    });
+
+    await waitFor(() => Number.parseInt(readFileSync(grandchildPidFile, "utf8"), 10) > 0);
+    const grandchildPid = Number.parseInt(readFileSync(grandchildPidFile, "utf8"), 10);
+    expect(isAlive(grandchildPid)).toBe(true);
+
+    await __terminateSupervisedChildrenForTests("cascade");
+    await child.waitExit();
+    await waitFor(() => !isAlive(grandchildPid));
+  });
+
+  it("escalates to SIGKILL after the grace period", async () => {
+    if (process.platform === "win32") {
+      return;
+    }
+
+    const child = superviseSpawn(process.execPath, [fixturePath, "keepalive"], {
+      stdio: "ignore",
+      killGraceMs: 50,
+      maxLifetimeMs: 5_000,
+    });
+
+    const realKill = process.kill.bind(process);
+    const processKillSpy = vi.spyOn(process, "kill").mockImplementation(((pid: number, signal?: NodeJS.Signals | number) => {
+      if (pid === -(child.pgid ?? 0)) {
+        return true;
+      }
+      return realKill(pid, signal as NodeJS.Signals | undefined);
+    }) as typeof process.kill);
+
+    await __terminateSupervisedChildrenForTests("sigkill");
+    expect(processKillSpy).toHaveBeenCalledWith(-(child.pgid ?? 0), "SIGTERM");
+    expect(processKillSpy).toHaveBeenCalledWith(-(child.pgid ?? 0), "SIGKILL");
+
+    processKillSpy.mockRestore();
+    child.child.kill("SIGKILL");
+    await expect(child.waitExit()).resolves.toEqual({ code: null, signal: "SIGKILL" });
+  });
+
+  it("enforces maxLifetimeMs", async () => {
+    const child = superviseSpawn(process.execPath, [fixturePath, "keepalive"], {
+      stdio: "ignore",
+      killGraceMs: 50,
+      maxLifetimeMs: 50,
+    });
+
+    const exit = await child.waitExit();
+    expect(exit.code === null || exit.code === 0 || exit.signal !== null).toBe(true);
+    await waitFor(() => __getProcessSupervisorStateForTests().registrySize === 0);
+  });
+
+  it("installs parent handlers only once", async () => {
+    const before = process.listenerCount("SIGTERM");
+
+    const first = superviseSpawn(process.execPath, [fixturePath, "exit-immediately"], { stdio: "ignore" });
+    const second = superviseSpawn(process.execPath, [fixturePath, "exit-immediately"], { stdio: "ignore" });
+
+    await Promise.all([first.waitExit(), second.waitExit()]);
+
+    expect(process.listenerCount("SIGTERM")).toBe(before + 1);
+    expect(__getProcessSupervisorStateForTests().handlersInstalled).toBe(true);
+  });
+
+  it("uses the Windows fallback branch when process groups are unavailable", async () => {
+    vi.spyOn(process, "platform", "get").mockReturnValue("win32");
+
+    const child = superviseSpawn(process.execPath, [fixturePath, "exit-immediately"], {
+      stdio: "ignore",
+      maxLifetimeMs: 100,
+    });
+
+    expect(child.pgid).toBeNull();
+    await expect(child.waitExit()).resolves.toEqual({ code: 0, signal: null });
+  });
+});

packages/core/src/index.ts

@@ -209,6 +209,12 @@ export {
 } from "./distributed-task-id.js";
 export type { DistributedTaskIdAllocator } from "./distributed-task-id.js";
 export { Database, createDatabase, toJson, toJsonNullable, fromJson } from "./db.js";
+export { ProcessSupervisor, superviseSpawn } from "./process-supervisor.js";
+export type {
+  SuperviseSpawnOptions,
+  SupervisedChild,
+  SupervisedExit,
+} from "./process-supervisor.js";
 export { DatabaseSync } from "./sqlite-adapter.js";
 export type { Statement, VacuumResult } from "./db.js";
 export { ArchiveDatabase } from "./archive-db.js";

packages/core/src/process-supervisor.ts

@@ -0,0 +1,335 @@
+import { spawn, type ChildProcess, type SpawnOptions } from "node:child_process";
+import { createLogger } from "./logger.js";
+
+const log = createLogger("process-supervisor");
+const DEFAULT_KILL_GRACE_MS = 2_000;
+const DEFAULT_MAX_LIFETIME_MS = 600_000;
+const MAX_KILL_WAIT_MS = 1_000;
+
+type ShutdownReason =
+  | { kind: "signal"; signal: NodeJS.Signals }
+  | { kind: "fatal"; source: "uncaughtException" | "unhandledRejection"; error: unknown }
+  | { kind: "exit"; code: number }
+  | { kind: "lifetime"; pid: number }
+  | { kind: "test"; label: string };
+
+export interface SuperviseSpawnOptions extends Omit<SpawnOptions, "detached"> {
+  /**
+   * Grace period between SIGTERM and SIGKILL when the supervisor tears a child
+   * down because the parent is exiting or a lifetime limit expires.
+   */
+  killGraceMs?: number;
+  /**
+   * Maximum time a supervised child may live before the supervisor forces it
+   * down. The timer is `unref()`'d so it never keeps the parent process alive.
+   */
+  maxLifetimeMs?: number;
+}
+
+export interface SupervisedExit {
+  code: number | null;
+  signal: NodeJS.Signals | null;
+}
+
+export interface SupervisedChild {
+  pid: number | undefined;
+  /**
+   * POSIX process-group id (same as child pid when `detached: true`).
+   * Windows cannot target negative PIDs, so `pgid` is `null` there.
+   */
+  pgid: number | null;
+  child: ChildProcess;
+  kill(signal?: NodeJS.Signals): void;
+  waitExit(): Promise<SupervisedExit>;
+}
+
+interface RegistryEntry {
+  child: ChildProcess;
+  pid: number | undefined;
+  pgid: number | null;
+  killGraceMs: number;
+  waitExit: Promise<SupervisedExit>;
+  lifetimeTimer: NodeJS.Timeout | null;
+  settled: boolean;
+  closeResult: SupervisedExit | null;
+}
+
+const registry = new Map<number, RegistryEntry>();
+let handlersInstalled = false;
+let activeShutdown: Promise<void> | null = null;
+const cleanupHandlers = new Map<string, (...args: unknown[]) => void>();
+
+function currentPlatform(): NodeJS.Platform {
+  return process.platform;
+}
+
+function usesProcessGroup(platform = currentPlatform()): boolean {
+  return platform !== "win32";
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function formatReason(reason: ShutdownReason): string {
+  switch (reason.kind) {
+    case "signal":
+      return reason.signal;
+    case "fatal":
+      return reason.source;
+    case "exit":
+      return `exit:${reason.code}`;
+    case "lifetime":
+      return `maxLifetime:${reason.pid}`;
+    case "test":
+      return `test:${reason.label}`;
+  }
+}
+
+function clearLifetimeTimer(entry: RegistryEntry): void {
+  if (entry.lifetimeTimer) {
+    clearTimeout(entry.lifetimeTimer);
+    entry.lifetimeTimer = null;
+  }
+}
+
+function deregister(entry: RegistryEntry, result: SupervisedExit): void {
+  if (entry.settled) {
+    return;
+  }
+  entry.settled = true;
+  entry.closeResult = result;
+  clearLifetimeTimer(entry);
+  if (typeof entry.pid === "number") {
+    registry.delete(entry.pid);
+    log.log(`child pid=${entry.pid} exited naturally code=${result.code ?? "null"} signal=${result.signal ?? "null"}`);
+  }
+}
+
+function killEntry(entry: RegistryEntry, signal: NodeJS.Signals = "SIGTERM"): void {
+  if (typeof entry.pid !== "number") {
+    return;
+  }
+
+  try {
+    if (entry.pgid !== null && usesProcessGroup()) {
+      process.kill(-entry.pgid, signal);
+      return;
+    }
+    entry.child.kill(signal);
+  } catch {
+    // Child or process group may already be gone.
+  }
+}
+
+async function terminateEntry(entry: RegistryEntry, reason: ShutdownReason): Promise<void> {
+  if (entry.settled) {
+    return;
+  }
+
+  log.warn(`terminating pid=${entry.pid ?? "unknown"} pgid=${entry.pgid ?? "n/a"} reason=${formatReason(reason)}`);
+  killEntry(entry, "SIGTERM");
+
+  const exitedWithinGrace = await Promise.race([
+    entry.waitExit.then(() => true),
+    sleep(entry.killGraceMs).then(() => false),
+  ]);
+
+  if (exitedWithinGrace || entry.settled) {
+    return;
+  }
+
+  log.warn(`grace expired for pid=${entry.pid ?? "unknown"}; escalating to SIGKILL`);
+  killEntry(entry, "SIGKILL");
+  log.warn(`sent SIGKILL to pid=${entry.pid ?? "unknown"} pgid=${entry.pgid ?? "n/a"}`);
+  await Promise.race([entry.waitExit, sleep(MAX_KILL_WAIT_MS)]);
+}
+
+async function terminateAll(reason: ShutdownReason): Promise<void> {
+  if (registry.size === 0) {
+    return;
+  }
+
+  if (!activeShutdown) {
+    activeShutdown = Promise.allSettled(
+      [...registry.values()].map((entry) => terminateEntry(entry, reason)),
+    ).then(() => undefined).finally(() => {
+      activeShutdown = null;
+    });
+  }
+
+  await activeShutdown;
+}
+
+function installHandlers(): void {
+  if (handlersInstalled) {
+    return;
+  }
+  handlersInstalled = true;
+
+  const onExit = (code: number) => {
+    for (const entry of registry.values()) {
+      killEntry(entry, "SIGTERM");
+    }
+    void code;
+  };
+
+  const makeSignalHandler = (signal: NodeJS.Signals) => {
+    const handler = () => {
+      void terminateAll({ kind: "signal", signal }).finally(() => {
+        const listener = cleanupHandlers.get(signal);
+        if (listener) {
+          process.removeListener(signal, listener as () => void);
+        }
+        process.kill(process.pid, signal);
+      });
+    };
+    return handler;
+  };
+
+  const handleFatal = (source: "uncaughtException" | "unhandledRejection", error: unknown) => {
+    void terminateAll({ kind: "fatal", source, error }).finally(() => {
+      const listener = cleanupHandlers.get(source);
+      if (listener) {
+        process.removeListener(source, listener as (value: unknown) => void);
+      }
+      if (source === "uncaughtException") {
+        throw error instanceof Error ? error : new Error(String(error));
+      }
+      throw error instanceof Error ? error : new Error(`Unhandled rejection: ${String(error)}`);
+    });
+  };
+
+  cleanupHandlers.set("exit", onExit as (...args: unknown[]) => void);
+  process.on("exit", onExit);
+
+  for (const signal of ["SIGINT", "SIGTERM", "SIGHUP"] as const) {
+    const handler = makeSignalHandler(signal);
+    cleanupHandlers.set(signal, handler as (...args: unknown[]) => void);
+    process.on(signal, handler);
+  }
+
+  const uncaughtHandler = (error: unknown) => {
+    handleFatal("uncaughtException", error);
+  };
+  cleanupHandlers.set("uncaughtException", uncaughtHandler as (...args: unknown[]) => void);
+  process.on("uncaughtException", uncaughtHandler);
+
+  const rejectionHandler = (reason: unknown) => {
+    handleFatal("unhandledRejection", reason);
+  };
+  cleanupHandlers.set("unhandledRejection", rejectionHandler as (...args: unknown[]) => void);
+  process.on("unhandledRejection", rejectionHandler);
+}
+
+/**
+ * Spawn a child process under parent-death supervision.
+ *
+ * On POSIX, the child is spawned with `detached: true`, which makes it the
+ * leader of a new process group. That lets the supervisor tear down the full
+ * subtree via `process.kill(-pgid, signal)` when the parent exits, receives a
+ * termination signal, throws an uncaught error, or hits `maxLifetimeMs`.
+ *
+ * On Windows, Node cannot signal a negative PID process group, so this falls
+ * back to a normal attached spawn plus direct `child.kill(signal)` tracking.
+ * That still reaps the immediate child on parent shutdown, but grandchildren
+ * are subject to platform limitations unless the child cooperatively forwards
+ * termination.
+ */
+export function superviseSpawn(
+  command: string,
+  args: readonly string[] = [],
+  options: SuperviseSpawnOptions = {},
+): SupervisedChild {
+  installHandlers();
+
+  const {
+    killGraceMs = DEFAULT_KILL_GRACE_MS,
+    maxLifetimeMs = DEFAULT_MAX_LIFETIME_MS,
+    ...spawnOptions
+  } = options;
+
+  const processGroup = usesProcessGroup();
+  const child = spawn(command, [...args], {
+    ...spawnOptions,
+    detached: processGroup,
+  });
+
+  let resolveExit: ((result: SupervisedExit) => void) | null = null;
+  const waitExit = new Promise<SupervisedExit>((resolve) => {
+    resolveExit = resolve;
+  });
+
+  const entry: RegistryEntry = {
+    child,
+    pid: child.pid,
+    pgid: processGroup && typeof child.pid === "number" ? child.pid : null,
+    killGraceMs,
+    waitExit,
+    lifetimeTimer: null,
+    settled: false,
+    closeResult: null,
+  };
+
+  child.once("close", (code, signal) => {
+    const result = { code, signal };
+    deregister(entry, result);
+    resolveExit?.(result);
+  });
+
+  if (typeof child.pid === "number") {
+    registry.set(child.pid, entry);
+    log.log(`spawned pid=${child.pid} pgid=${entry.pgid ?? "n/a"} command=${command}`);
+  } else {
+    log.warn(`spawned child without pid for command=${command}`);
+  }
+
+  if (Number.isFinite(maxLifetimeMs) && maxLifetimeMs > 0) {
+    entry.lifetimeTimer = setTimeout(() => {
+      log.warn(`maxLifetime exceeded for pid=${entry.pid ?? "unknown"} after ${maxLifetimeMs}ms`);
+      void terminateEntry(entry, { kind: "lifetime", pid: entry.pid ?? -1 });
+    }, maxLifetimeMs);
+    entry.lifetimeTimer.unref();
+  }
+
+  return {
+    pid: child.pid,
+    pgid: entry.pgid,
+    child,
+    kill(signal = "SIGTERM") {
+      killEntry(entry, signal);
+    },
+    waitExit() {
+      return waitExit;
+    },
+  };
+}
+
+export const ProcessSupervisor = {
+  superviseSpawn,
+} as const;
+
+export function __getProcessSupervisorStateForTests(): { registrySize: number; handlersInstalled: boolean } {
+  return {
+    registrySize: registry.size,
+    handlersInstalled,
+  };
+}
+
+export async function __terminateSupervisedChildrenForTests(label = "test"): Promise<void> {
+  await terminateAll({ kind: "test", label });
+}
+
+export function __resetProcessSupervisorForTests(): void {
+  for (const entry of registry.values()) {
+    clearLifetimeTimer(entry);
+    killEntry(entry, "SIGKILL");
+  }
+  registry.clear();
+  activeShutdown = null;
+  for (const [event, handler] of cleanupHandlers.entries()) {
+    process.removeListener(event as NodeJS.Signals | "uncaughtException" | "unhandledRejection" | "exit", handler as (...args: unknown[]) => void);
+  }
+  cleanupHandlers.clear();
+  handlersInstalled = false;
+}