Recover stale supervised daemons

Teach daemon stop to use lifecycle shutdown when the API is reachable even if the owner pid is stale, then wait for the API to disappear and clean the stale pidfile. Launch desktop-managed daemons detached from desktop stdio, preserve stale reachable daemon ownership for version checks, and allow desktop stop/restart to use the CLI recovery path. Add supervisor heartbeats so supervised workers shut down when their supervisor disappears instead of surviving as orphaned reachable daemons.
2026-07-29 12:01:31 +00:00 · 2026-06-07 23:18:48 +07:00
parent 7c8b290e2f
commit 893e3376b0
7 changed files with 769 additions and 115 deletions
--- a/packages/cli/src/commands/daemon/local-daemon.ts
+++ b/packages/cli/src/commands/daemon/local-daemon.ts
@@ -1,5 +1,5 @@
 import { spawnSync, type ChildProcess } from "node:child_process";
-import { existsSync, readFileSync } from "node:fs";
+import { existsSync, readFileSync, unlinkSync } from "node:fs";
 import { createRequire } from "node:module";
 import path from "node:path";
 import { loadConfig, resolvePaseoHome, spawnProcess } from "@getpaseo/server";
@@ -351,6 +351,123 @@ async function waitForPidExit(pid: number, timeoutMs: number): Promise<boolean>
  return poll();
 }

+async function waitForDaemonUnreachable(
+  state: LocalDaemonState,
+  timeoutMs: number,
+): Promise<boolean> {
+  const host = resolveTcpHostFromListen(state.listen);
+  if (!host) {
+    return true;
+  }
+
+  const reachableHost = host;
+  const deadline = Date.now() + timeoutMs;
+  async function poll(): Promise<boolean> {
+    const client = await tryConnectToDaemon({ host: reachableHost, timeout: 500 });
+    if (!client) {
+      return true;
+    }
+    await client.close().catch(() => undefined);
+    if (Date.now() >= deadline) {
+      const finalClient = await tryConnectToDaemon({
+        host: reachableHost,
+        timeout: PID_POLL_INTERVAL_MS,
+      });
+      if (!finalClient) {
+        return true;
+      }
+      await finalClient.close().catch(() => undefined);
+      return false;
+    }
+    await sleep(PID_POLL_INTERVAL_MS);
+    return poll();
+  }
+
+  return poll();
+}
+
+function removeStalePidFile(state: LocalDaemonState): void {
+  if (!state.stalePidFile) {
+    return;
+  }
+
+  try {
+    unlinkSync(state.pidPath);
+  } catch {
+    // Best-effort cleanup only. The successful lifecycle stop is authoritative.
+  }
+}
+
+function createNotRunningStopResult(
+  state: LocalDaemonState,
+  pid: number | null,
+  message: string,
+): StopLocalDaemonResult {
+  return {
+    action: "not_running",
+    home: state.home,
+    pid,
+    forced: false,
+    message,
+  };
+}
+
+function createStopTimeoutError(
+  state: LocalDaemonState,
+  pid: number | null,
+  timeoutMs: number,
+): Error {
+  if (!state.running) {
+    const host = resolveTcpHostFromListen(state.listen);
+    return new Error(
+      `Timed out waiting for daemon${host ? ` at ${host}` : ""} to stop after ${Math.ceil(
+        timeoutMs / 1000,
+      )}s`,
+    );
+  }
+  return new Error(
+    `Timed out waiting for daemon PID ${pid} to stop after ${Math.ceil(timeoutMs / 1000)}s`,
+  );
+}
+
+async function signalDaemonOwnerForStop(
+  state: LocalDaemonState,
+  pid: number | null,
+): Promise<StopLocalDaemonResult | null> {
+  if (pid === null) {
+    return createNotRunningStopResult(state, null, "Daemon is not running");
+  }
+
+  const signaled = await signalProcessTreeOrOwnerSafely(pid, "SIGTERM");
+  if (signaled) {
+    return null;
+  }
+
+  return createNotRunningStopResult(state, pid, "Daemon process was already stopped");
+}
+
+async function waitForStopAfterRequest(args: {
+  state: LocalDaemonState;
+  pid: number | null;
+  timeoutMs: number;
+  killTimeoutMs: number;
+  force?: boolean;
+}): Promise<{ stopped: boolean; forced: boolean }> {
+  const { state, pid, timeoutMs, killTimeoutMs, force } = args;
+  let stopped =
+    state.running && pid !== null
+      ? await waitForPidExit(pid, timeoutMs)
+      : await waitForDaemonUnreachable(state, timeoutMs);
+
+  if (!stopped && force && state.running && pid !== null) {
+    await signalProcessTreeOrOwnerSafely(pid, "SIGKILL");
+    stopped = await waitForPidExit(pid, killTimeoutMs);
+    return { stopped, forced: true };
+  }
+
+  return { stopped, forced: false };
+}
+
 type LifecycleShutdownAttempt = { requested: true } | { requested: false; reason: string };

 function getErrorMessage(error: unknown): string {
@@ -567,49 +684,41 @@ export async function stopLocalDaemon(
  const killTimeoutMs = options.killTimeoutMs ?? DEFAULT_KILL_TIMEOUT_MS;
  const state = resolveLocalDaemonState({ home: options.home });

-  if (!state.pidInfo || !state.running) {
-    const staleSuffix =
-      state.stalePidFile && state.pidInfo ? ` (stale PID file for ${state.pidInfo.pid})` : "";
-    return {
-      action: "not_running",
-      home: state.home,
-      pid: state.pidInfo?.pid ?? null,
-      forced: false,
-      message: `Daemon is not running${staleSuffix}`,
-    };
-  }
-
-  const pid = state.pidInfo.pid;
  const shutdownAttempt = await requestLifecycleShutdown(state, timeoutMs);
  const lifecycleRequested = shutdownAttempt.requested;
-  const fallbackMessage = shutdownAttempt.requested ? null : shutdownAttempt.reason;
-  let forced = false;
-  if (!lifecycleRequested) {
-    const signaled = await signalProcessTreeOrOwnerSafely(pid, "SIGTERM");
-    if (!signaled) {
-      return {
-        action: "not_running",
-        home: state.home,
-        pid,
-        forced: false,
-        message: "Daemon process was already stopped",
-      };
-    }
-  }

-  let stopped = await waitForPidExit(pid, timeoutMs);
-  if (!stopped && options.force) {
-    forced = true;
-    await signalProcessTreeOrOwnerSafely(pid, "SIGKILL");
-    stopped = await waitForPidExit(pid, killTimeoutMs);
-  }
-
-  if (!stopped) {
-    throw new Error(
-      `Timed out waiting for daemon PID ${pid} to stop after ${Math.ceil(timeoutMs / 1000)}s`,
+  if (!state.pidInfo || (!state.running && !lifecycleRequested)) {
+    const staleSuffix =
+      state.stalePidFile && state.pidInfo ? ` (stale PID file for ${state.pidInfo.pid})` : "";
+    return createNotRunningStopResult(
+      state,
+      state.pidInfo?.pid ?? null,
+      `Daemon is not running${staleSuffix}`,
    );
  }

+  const pid = state.pidInfo?.pid ?? null;
+  const fallbackMessage = shutdownAttempt.requested ? null : shutdownAttempt.reason;
+  if (!lifecycleRequested) {
+    const notRunningResult = await signalDaemonOwnerForStop(state, pid);
+    if (notRunningResult) return notRunningResult;
+  }
+
+  const { stopped, forced } = await waitForStopAfterRequest({
+    state,
+    pid,
+    timeoutMs,
+    killTimeoutMs,
+    force: options.force,
+  });
+  if (!stopped) {
+    throw createStopTimeoutError(state, pid, timeoutMs);
+  }
+
+  if (lifecycleRequested) {
+    removeStalePidFile(state);
+  }
+
  return {
    action: "stopped",
    home: state.home,
--- a/packages/cli/tests/34-daemon-stop-stale-reachable.test.ts
+++ b/packages/cli/tests/34-daemon-stop-stale-reachable.test.ts
@@ -0,0 +1,207 @@
+#!/usr/bin/env npx tsx
+
+/**
+ * Regression: `paseo daemon stop` must stop a reachable daemon even when the
+ * local pid file points at a dead supervisor owner.
+ */
+
+import assert from "node:assert";
+import { spawn, type ChildProcess } from "node:child_process";
+import { existsSync } from "node:fs";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { $ } from "zx";
+import { getAvailablePort } from "./helpers/network.ts";
+
+$.verbose = false;
+
+const pollIntervalMs = 100;
+const testEnv = {
+  PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD: process.env.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD ?? "0",
+  PASEO_DICTATION_ENABLED: process.env.PASEO_DICTATION_ENABLED ?? "0",
+  PASEO_VOICE_MODE_ENABLED: process.env.PASEO_VOICE_MODE_ENABLED ?? "0",
+};
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function isProcessRunning(pid: number): boolean {
+  if (!Number.isInteger(pid) || pid <= 0) {
+    return false;
+  }
+
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+async function waitFor(
+  check: () => Promise<boolean> | boolean,
+  timeoutMs: number,
+  message: string,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+
+  async function poll(): Promise<void> {
+    if (await check()) return;
+    if (Date.now() >= deadline) throw new Error(message);
+    await sleep(pollIntervalMs);
+    return poll();
+  }
+
+  return poll();
+}
+
+interface DaemonStatus {
+  localDaemon: string | null;
+  connectedDaemon: string | null;
+  pid: number | null;
+}
+
+async function readDaemonStatus(paseoHome: string): Promise<DaemonStatus> {
+  const result =
+    await $`PASEO_HOME=${paseoHome} PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD=${testEnv.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD} PASEO_DICTATION_ENABLED=${testEnv.PASEO_DICTATION_ENABLED} PASEO_VOICE_MODE_ENABLED=${testEnv.PASEO_VOICE_MODE_ENABLED} npx paseo daemon status --home ${paseoHome} --json`.nothrow();
+  if (result.exitCode !== 0) {
+    return { localDaemon: null, connectedDaemon: null, pid: null };
+  }
+
+  try {
+    const parsed = JSON.parse(result.stdout) as {
+      localDaemon?: unknown;
+      connectedDaemon?: unknown;
+      pid?: unknown;
+    };
+    return {
+      localDaemon: typeof parsed.localDaemon === "string" ? parsed.localDaemon : null,
+      connectedDaemon: typeof parsed.connectedDaemon === "string" ? parsed.connectedDaemon : null,
+      pid:
+        typeof parsed.pid === "number" && Number.isInteger(parsed.pid) && parsed.pid > 0
+          ? parsed.pid
+          : null,
+    };
+  } catch {
+    return { localDaemon: null, connectedDaemon: null, pid: null };
+  }
+}
+
+function findUnusedPid(): number {
+  for (let pid = 999_999; pid > 900_000; pid--) {
+    if (!isProcessRunning(pid)) {
+      return pid;
+    }
+  }
+  throw new Error("Unable to find unused pid for stale pid fixture");
+}
+
+console.log("=== Daemon Stop (stale pid, reachable worker regression) ===\n");
+
+const port = await getAvailablePort();
+const paseoHome = await mkdtemp(join(tmpdir(), "paseo-stop-stale-reachable-"));
+const cliRoot = join(import.meta.dirname, "..");
+const host = `127.0.0.1:${port}`;
+const pidPath = join(paseoHome, "paseo.pid");
+const stalePid = findUnusedPid();
+
+let workerProcess: ChildProcess | null = null;
+
+try {
+  console.log("Test 1: start daemon worker with stale supervisor pid file");
+
+  await writeFile(
+    pidPath,
+    `${JSON.stringify(
+      {
+        pid: stalePid,
+        startedAt: new Date().toISOString(),
+        hostname: "stale-supervisor-fixture.local",
+        uid: typeof process.getuid === "function" ? process.getuid() : undefined,
+        listen: host,
+      },
+      null,
+      2,
+    )}\n`,
+  );
+
+  workerProcess = spawn(
+    process.execPath,
+    ["--import", "tsx", "../server/src/server/daemon-worker.ts"],
+    {
+      cwd: cliRoot,
+      env: {
+        ...process.env,
+        ...testEnv,
+        PASEO_HOME: paseoHome,
+        PASEO_LISTEN: host,
+        PASEO_RELAY_ENABLED: "false",
+        CI: "true",
+      },
+      stdio: ["ignore", "pipe", "pipe"],
+    },
+  );
+
+  await waitFor(
+    async () => {
+      const status = await readDaemonStatus(paseoHome);
+      return status.localDaemon === "stale_pid" && status.connectedDaemon === "reachable";
+    },
+    120000,
+    "daemon did not enter stale_pid + reachable state in time",
+  );
+
+  const statusBeforeStop = await readDaemonStatus(paseoHome);
+  assert.strictEqual(statusBeforeStop.pid, stalePid, "status should report the stale owner pid");
+  assert(workerProcess.pid && isProcessRunning(workerProcess.pid), "worker should be running");
+  console.log(`✓ fixture has stale pid ${stalePid} and live worker ${workerProcess.pid}\n`);
+
+  console.log(
+    "Test 2: `paseo daemon stop` should stop reachable worker instead of saying not_running",
+  );
+  const stopResult =
+    await $`PASEO_HOME=${paseoHome} PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD=${testEnv.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD} PASEO_DICTATION_ENABLED=${testEnv.PASEO_DICTATION_ENABLED} PASEO_VOICE_MODE_ENABLED=${testEnv.PASEO_VOICE_MODE_ENABLED} npx paseo daemon stop --home ${paseoHome} --json`.nothrow();
+  assert.strictEqual(stopResult.exitCode, 0, `stop should succeed: ${stopResult.stderr}`);
+  const stopJson = JSON.parse(stopResult.stdout) as {
+    action?: unknown;
+    pid?: unknown;
+    message?: unknown;
+  };
+  assert.strictEqual(stopJson.action, "stopped", "stop should report stopped action");
+  assert.strictEqual(
+    stopJson.pid,
+    String(stalePid),
+    "stop should report the stale pid it recovered from",
+  );
+  assert.strictEqual(
+    stopJson.message,
+    "Daemon stopped gracefully",
+    "stop should route through lifecycle shutdown",
+  );
+
+  await waitFor(
+    () => !isProcessRunning(workerProcess?.pid ?? -1),
+    15000,
+    "worker remained running after stop",
+  );
+  assert.strictEqual(existsSync(pidPath), false, "stale pid file should be removed after stop");
+  console.log("✓ stop recovered stale supervisor pid state\n");
+} finally {
+  if (workerProcess?.pid && isProcessRunning(workerProcess.pid)) {
+    workerProcess.kill("SIGTERM");
+    await waitFor(
+      () => !isProcessRunning(workerProcess!.pid ?? -1),
+      5000,
+      "worker cleanup timed out",
+    ).catch(() => {
+      workerProcess?.kill("SIGKILL");
+    });
+  }
+
+  await $`PASEO_HOME=${paseoHome} PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD=${testEnv.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD} PASEO_DICTATION_ENABLED=${testEnv.PASEO_DICTATION_ENABLED} PASEO_VOICE_MODE_ENABLED=${testEnv.PASEO_VOICE_MODE_ENABLED} npx paseo daemon stop --home ${paseoHome} --force`.nothrow();
+  await rm(paseoHome, { recursive: true, force: true });
+}
+
+console.log("=== Stale reachable stop regression test passed ===");
--- a/packages/cli/tests/35-daemon-worker-supervisor-disconnect.test.ts
+++ b/packages/cli/tests/35-daemon-worker-supervisor-disconnect.test.ts
@@ -0,0 +1,192 @@
+#!/usr/bin/env npx tsx
+
+/**
+ * Regression: a supervised daemon worker must exit when its supervisor IPC
+ * channel closes, instead of becoming an orphaned daemon.
+ */
+
+import assert from "node:assert";
+import { spawn, spawnSync, type ChildProcess } from "node:child_process";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { $ } from "zx";
+import { getAvailablePort } from "./helpers/network.ts";
+
+$.verbose = false;
+
+const pollIntervalMs = 100;
+const testEnv = {
+  PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD: process.env.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD ?? "0",
+  PASEO_DICTATION_ENABLED: process.env.PASEO_DICTATION_ENABLED ?? "0",
+  PASEO_VOICE_MODE_ENABLED: process.env.PASEO_VOICE_MODE_ENABLED ?? "0",
+};
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function isProcessRunning(pid: number): boolean {
+  if (!Number.isInteger(pid) || pid <= 0) {
+    return false;
+  }
+
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+function readWorkerPid(supervisorPid: number): number | null {
+  if (!Number.isInteger(supervisorPid) || supervisorPid <= 0) {
+    return null;
+  }
+
+  const result = spawnSync("ps", ["ax", "-o", "pid=,ppid="], { encoding: "utf8" });
+  if (result.status !== 0 || result.error) {
+    return null;
+  }
+
+  for (const line of result.stdout.split("\n")) {
+    const trimmed = line.trim();
+    if (!trimmed) {
+      continue;
+    }
+    const [pidToken, ppidToken] = trimmed.split(/\s+/);
+    const pid = Number.parseInt(pidToken ?? "", 10);
+    const ppid = Number.parseInt(ppidToken ?? "", 10);
+    if (ppid === supervisorPid && pid > 0) {
+      return pid;
+    }
+  }
+
+  return null;
+}
+
+async function waitFor(
+  check: () => Promise<boolean> | boolean,
+  timeoutMs: number,
+  message: string,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+
+  async function poll(): Promise<void> {
+    if (await check()) return;
+    if (Date.now() >= deadline) throw new Error(message);
+    await sleep(pollIntervalMs);
+    return poll();
+  }
+
+  return poll();
+}
+
+interface DaemonStatus {
+  localDaemon: string | null;
+  pid: number | null;
+}
+
+async function readDaemonStatus(paseoHome: string): Promise<DaemonStatus> {
+  const result =
+    await $`PASEO_HOME=${paseoHome} PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD=${testEnv.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD} PASEO_DICTATION_ENABLED=${testEnv.PASEO_DICTATION_ENABLED} PASEO_VOICE_MODE_ENABLED=${testEnv.PASEO_VOICE_MODE_ENABLED} npx paseo daemon status --home ${paseoHome} --json`.nothrow();
+  if (result.exitCode !== 0) {
+    return { localDaemon: null, pid: null };
+  }
+
+  try {
+    const parsed = JSON.parse(result.stdout) as { localDaemon?: unknown; pid?: unknown };
+    return {
+      localDaemon: typeof parsed.localDaemon === "string" ? parsed.localDaemon : null,
+      pid:
+        typeof parsed.pid === "number" && Number.isInteger(parsed.pid) && parsed.pid > 0
+          ? parsed.pid
+          : null,
+    };
+  } catch {
+    return { localDaemon: null, pid: null };
+  }
+}
+
+console.log("=== Daemon Worker Supervisor Disconnect Regression ===\n");
+
+const port = await getAvailablePort();
+const paseoHome = await mkdtemp(join(tmpdir(), "paseo-worker-supervisor-disconnect-"));
+const cliRoot = join(import.meta.dirname, "..");
+
+let supervisorProcess: ChildProcess | null = null;
+let recentSupervisorLogs = "";
+
+try {
+  console.log("Test 1: start supervised daemon with isolated PASEO_HOME");
+
+  supervisorProcess = spawn(
+    process.execPath,
+    ["--import", "tsx", "../server/scripts/supervisor-entrypoint.ts", "--dev"],
+    {
+      cwd: cliRoot,
+      env: {
+        ...process.env,
+        ...testEnv,
+        PASEO_HOME: paseoHome,
+        PASEO_LISTEN: `127.0.0.1:${port}`,
+        PASEO_RELAY_ENABLED: "false",
+        CI: "true",
+      },
+      stdio: ["ignore", "pipe", "pipe"],
+    },
+  );
+
+  supervisorProcess.stdout?.on("data", (chunk) => {
+    recentSupervisorLogs = (recentSupervisorLogs + chunk.toString()).slice(-8000);
+  });
+  supervisorProcess.stderr?.on("data", (chunk) => {
+    recentSupervisorLogs = (recentSupervisorLogs + chunk.toString()).slice(-8000);
+  });
+
+  await waitFor(
+    async () => {
+      const status = await readDaemonStatus(paseoHome);
+      return (
+        status.localDaemon === "running" && status.pid !== null && isProcessRunning(status.pid)
+      );
+    },
+    120000,
+    "daemon did not become running in time",
+  );
+
+  const statusBeforeKill = await readDaemonStatus(paseoHome);
+  const supervisorPid = statusBeforeKill.pid;
+  assert(supervisorPid !== null, "supervisor pid should exist once daemon starts");
+  const workerPid = readWorkerPid(supervisorPid);
+  assert(workerPid !== null, "supervisor should have a worker process");
+  assert(isProcessRunning(workerPid), "worker process should be running");
+  console.log(`✓ daemon running with supervisor ${supervisorPid} and worker ${workerPid}\n`);
+
+  console.log("Test 2: killing supervisor should make worker exit via IPC disconnect");
+  supervisorProcess.kill("SIGKILL");
+  await waitFor(
+    () => !isProcessRunning(supervisorPid),
+    15000,
+    "supervisor remained running after SIGKILL",
+  );
+  await waitFor(
+    () => !isProcessRunning(workerPid),
+    15000,
+    "worker remained running after supervisor IPC disconnect",
+  );
+  console.log("✓ worker exited after supervisor disconnect\n");
+} finally {
+  if (supervisorProcess?.pid && isProcessRunning(supervisorProcess.pid)) {
+    supervisorProcess.kill("SIGKILL");
+  }
+
+  await $`PASEO_HOME=${paseoHome} PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD=${testEnv.PASEO_LOCAL_SPEECH_AUTO_DOWNLOAD} PASEO_DICTATION_ENABLED=${testEnv.PASEO_DICTATION_ENABLED} PASEO_VOICE_MODE_ENABLED=${testEnv.PASEO_VOICE_MODE_ENABLED} npx paseo daemon stop --home ${paseoHome} --force`.nothrow();
+  await rm(paseoHome, { recursive: true, force: true });
+}
+
+if (recentSupervisorLogs.trim().length === 0) {
+  console.log("(no supervisor logs captured)");
+}
+
+console.log("=== Worker supervisor disconnect regression test passed ===");