fix(daemon): keep lifecycle operations on the selected instance

Restart reconstructed launch inputs and address fallback could redirect a home-selected shutdown to another daemon. Let supervisors publish bound endpoints, retain their worker restart contract, and share captured local lifecycle ownership between CLI and Desktop.
This commit is contained in:
Mohamed Boudra
2026-09-09 14:09:53 +02:00
parent fdf3b4b47f
commit 264276880e
166 changed files with 3824 additions and 4467 deletions
@@ -18,13 +18,11 @@ process.title = "Paseo Supervisor";
interface DaemonRunnerConfig {
devMode: boolean;
reclaimStalePidLock: boolean;
workerArgs: string[];
}
function parseConfig(argv: string[]): DaemonRunnerConfig {
let devMode = false;
let reclaimStalePidLock = false;
const workerArgs: string[] = [];
for (const arg of argv) {
@@ -33,13 +31,14 @@ function parseConfig(argv: string[]): DaemonRunnerConfig {
continue;
}
if (arg === "--reclaim-stale-pid-lock") {
reclaimStalePidLock = true;
continue;
throw new Error(
"--reclaim-stale-pid-lock was removed: stop the existing supervisor before starting another.",
);
}
workerArgs.push(arg);
}
return { devMode, reclaimStalePidLock, workerArgs };
return { devMode, workerArgs };
}
function resolveWorkerEntry(): string {
@@ -116,7 +115,6 @@ async function main(): Promise<void> {
try {
await acquirePidLock(paseoHome, null, {
ownerPid: process.pid,
reclaimStaleDesktopLock: config.reclaimStalePidLock,
});
} catch (error) {
if (error instanceof PidLockError) {
@@ -177,6 +175,7 @@ async function main(): Promise<void> {
onWorkerReady: async ({ listen }) => {
await updatePidLock(paseoHome, { listen }, { ownerPid: process.pid });
},
onWorkerExit: () => updatePidLock(paseoHome, { listen: null }, { ownerPid: process.pid }),
onSupervisorExit: releaseLock,
});
requestSupervisorShutdown = supervisor.requestShutdown;
@@ -36,7 +36,10 @@ async function runSupervisorFixture(options: {
const workerPath = path.join(tempDir, "worker.mjs");
const runnerPath = path.join(tempDir, "runner.mjs");
await writeFile(workerPath, options.workerSource);
await writeFile(
workerPath,
`process.send?.({ type: "paseo:ready", listen: "fixture" });\n${options.workerSource}`,
);
await writeFile(
runnerPath,
`
@@ -0,0 +1,122 @@
import { spawn } from "node:child_process";
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { expect, test } from "vitest";
async function runWorker(worker: string, publicationDelay = 0) {
const home = await mkdtemp(path.join(tmpdir(), "paseo-readiness-"));
const workerPath = path.join(home, "worker.mjs");
const runnerPath = path.join(home, "runner.mjs");
const eventsPath = path.join(home, "events.jsonl");
const supervisor = new URL("./supervisor.ts", import.meta.url).href;
await writeFile(workerPath, worker);
await writeFile(
runnerPath,
`
import { appendFile } from "node:fs/promises";
import { runSupervisor } from ${JSON.stringify(supervisor)};
const record = (event) => appendFile(${JSON.stringify(eventsPath)}, JSON.stringify(event) + "\\n");
runSupervisor({
name: "ReadinessTest", startupMessage: "starting", restartOnCrash: true,
resolveWorkerEntry: () => ${JSON.stringify(workerPath)}, workerArgs: [], workerExecArgv: [],
onWorkerReady: async ({ listen }) => {
await new Promise(resolve => setTimeout(resolve, ${publicationDelay}));
await record(listen);
},
onWorkerExit: () => record(null),
});
`,
);
const env = Object.fromEntries(
Object.entries(process.env).filter(([key]) => !key.startsWith("PASEO_")),
);
const child = spawn(process.execPath, ["--import", "tsx", runnerPath], {
cwd: fileURLToPath(new URL("../../../", import.meta.url)),
env: { ...env, HOME: home, PASEO_HOME: home },
stdio: ["ignore", "pipe", "pipe"],
});
let output = "";
child.stdout.on("data", (chunk) => {
output += chunk;
});
child.stderr.on("data", (chunk) => {
output += chunk;
});
try {
const code = await new Promise<number | null>((resolve, reject) => {
const timer = setTimeout(() => {
child.kill("SIGKILL");
reject(new Error(output));
}, 5_000);
child.once("error", reject);
child.once("close", (exitCode) => {
clearTimeout(timer);
resolve(exitCode);
});
});
const events = await readFile(eventsPath, "utf8").catch(() => "");
return {
code,
output,
events: events
.trim()
.split("\n")
.filter(Boolean)
.map((line) => JSON.parse(line)),
};
} finally {
await rm(home, { recursive: true, force: true });
}
}
test("a worker that exits before ready fails the supervisor without respawning", async () => {
const result = await runWorker(`
import { existsSync, writeFileSync } from "node:fs";
const marker = process.argv[1] + ".started";
if (existsSync(marker)) process.exit(0);
writeFileSync(marker, "started");
process.exit(1);
`);
expect(result.code).toBe(1);
expect(result.output).not.toContain("Restarting worker");
});
test("a replacement worker must independently reach ready", async () => {
const result = await runWorker(`
import { existsSync, writeFileSync } from "node:fs";
const marker = process.argv[1] + ".started";
if (existsSync(marker)) process.exit(0);
writeFileSync(marker, "started");
process.on("message", message => {
if (message.type === "paseo:graceful-shutdown") process.exit(0);
});
process.send({ type: "paseo:ready", listen: "test-endpoint" });
process.send({ type: "paseo:restart" });
`);
expect(result.code).toBe(1);
expect(result.events, result.output).toEqual(["test-endpoint", null, null]);
});
test("exit clears publication after an in-flight ready write", async () => {
const result = await runWorker(
`
process.send({ type: "paseo:ready", listen: "test-endpoint" }, () => process.exit(0));
`,
100,
);
expect(result.code).toBe(0);
expect(result.events.at(-1)).toBeNull();
});
test("requested shutdown before first readiness exits successfully", async () => {
const result = await runWorker(`
process.on("message", message => {
if (message.type === "paseo:graceful-shutdown") process.exit(0);
});
process.send({ type: "paseo:shutdown", reason: "cancelled_start" });
`);
expect(result.code).toBe(0);
expect(result.events).toEqual([null]);
});
+63 -30
View File
@@ -51,6 +51,7 @@ interface SupervisorOptions {
env?: NodeJS.ProcessEnv;
} | null;
onWorkerReady?: (message: { listen: string }) => Promise<void> | void;
onWorkerExit?: () => Promise<void> | void;
restartOnCrash?: boolean;
onSupervisorExit?: () => Promise<void> | void;
logFile?: SupervisorLogFileOptions;
@@ -129,6 +130,8 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
let restarting = false;
let shuttingDown = false;
let exiting = false;
let lifecycleFailed = false;
let publication = Promise.resolve();
let forceKillTimer: NodeJS.Timeout | null = null;
const logStream = createSupervisorLogStream(options.logFile);
@@ -244,6 +247,8 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
}
const currentChild = child;
let reachedReady = false;
// Serialize endpoint writes with exit/clear before allowing another worker to spawn.
const heartbeat = setInterval(() => {
const message: SupervisorHeartbeatMessage = { type: "paseo:supervisor-heartbeat" };
if (currentChild.connected) {
@@ -276,18 +281,21 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
child.on("message", (msg: unknown) => {
const lifecycleMessage = parseLifecycleMessage(msg);
if (!lifecycleMessage) {
if (!lifecycleMessage || child !== currentChild) {
return;
}
if (lifecycleMessage.type === "paseo:ready") {
reachedReady = true;
writeLifecycleLog("Worker ready", { listen: lifecycleMessage.listen });
Promise.resolve(options.onWorkerReady?.({ listen: lifecycleMessage.listen })).catch(
(error) => {
publication = publication
.then(() => options.onWorkerReady?.({ listen: lifecycleMessage.listen }))
.catch((error) => {
lifecycleFailed = true;
const message = error instanceof Error ? error.message : String(error);
log(`Worker ready callback failed: ${message}`);
},
);
requestShutdown("endpoint_publication_failed");
});
return;
}
@@ -303,35 +311,57 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
requestRestart(reason);
});
child.on("error", (error) => {
lifecycleFailed = true;
log(`Worker spawn failed: ${error.message}`);
exitSupervisor(1);
});
child.on("exit", (code, signal) => {
clearInterval(heartbeat);
clearForceKillTimer();
const exitDescriptor = describeExit(code, signal);
writeLifecycleLog("Worker exited", { code, signal, exit: exitDescriptor });
child = null;
publication = publication
.then(() => options.onWorkerExit?.())
.catch((error) => {
lifecycleFailed = true;
log(`Worker exit callback failed: ${String(error)}`);
});
void publication.then(() => {
const exitDescriptor = describeExit(code, signal);
writeLifecycleLog("Worker exited", { code, signal, exit: exitDescriptor });
if (shuttingDown) {
log(`Worker exited (${exitDescriptor}). Supervisor shutting down.`);
exitSupervisor(0);
if (lifecycleFailed || (!reachedReady && !shuttingDown)) {
log(`Worker exited before successful readiness (${exitDescriptor}). Supervisor exiting.`);
exitSupervisor(1);
return;
}
if (shuttingDown) {
log(`Worker exited (${exitDescriptor}). Supervisor shutting down.`);
exitSupervisor(0);
return;
}
const crashed =
restartOnCrash &&
((code !== 0 && code !== null) || (signal !== null && signal !== "SIGTERM"));
if (restarting || crashed) {
restarting = false;
log(
crashed
? `Worker crashed (${exitDescriptor}). Restarting worker...`
: `Worker exited (${exitDescriptor}). Restarting worker...`,
);
spawnWorker();
return;
}
log(`Worker exited (${exitDescriptor}). Supervisor exiting.`);
exitSupervisor(typeof code === "number" ? code : 1);
return;
}
const crashed =
restartOnCrash &&
((code !== 0 && code !== null) || (signal !== null && signal !== "SIGTERM"));
if (restarting || crashed) {
restarting = false;
log(
crashed
? `Worker crashed (${exitDescriptor}). Restarting worker...`
: `Worker exited (${exitDescriptor}). Restarting worker...`,
);
spawnWorker();
return;
}
log(`Worker exited (${exitDescriptor}). Supervisor exiting.`);
exitSupervisor(typeof code === "number" ? code : 1);
});
});
};
@@ -389,7 +419,10 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
writeLifecycleLog("Supervisor shutdown requested", { reason });
log(`${reason}. Stopping worker...`);
if (!child) {
exitSupervisor(0);
void publication.then(() => {
exitSupervisor(lifecycleFailed ? 1 : 0);
return;
});
return;
}
requestWorkerShutdown(reason);
+2 -2
View File
@@ -80,12 +80,12 @@ export function parseListenString(listen: string): ListenTarget {
throw new Error(`Invalid listen string: ${listen}`);
}
function formatListenTarget(listenTarget: ListenTarget | null): string | null {
export function formatListenTarget(listenTarget: ListenTarget | null): string | null {
if (!listenTarget) {
return null;
}
if (listenTarget.type === "tcp") {
return `${listenTarget.host}:${listenTarget.port}`;
return `${formatHostForHttpUrl(listenTarget.host)}:${listenTarget.port}`;
}
return listenTarget.path;
}
@@ -0,0 +1,90 @@
// Daemon configuration inputs. General provider credentials and executable/runtime
// controls remain available to managed launches and their agent processes.
export const DAEMON_SETTING_ENV_KEYS = [
"MCP_DEBUG",
"OPENAI_STT_BASE_URL",
"OPENAI_TTS_BASE_URL",
"PASEO_ALLOWED_HOSTS",
"PASEO_APP_BASE_URL",
"PASEO_CORS_ORIGINS",
"PASEO_DICTATION_ENABLED",
"PASEO_DICTATION_LANGUAGE",
"PASEO_DICTATION_LOCAL_STT_MODEL",
"PASEO_DICTATION_STT_PROVIDER",
"PASEO_GIT_CONCURRENCY",
"PASEO_GIT_MAX_PROCESSES_PER_SECOND",
"PASEO_GIT_MAX_PROCESS_CONCURRENCY",
"PASEO_HOSTNAMES",
"PASEO_LISTEN",
"PASEO_LOCAL_MODELS_DIR",
"PASEO_LOG",
"PASEO_LOG_CONSOLE_FORMAT",
"PASEO_LOG_CONSOLE_LEVEL",
"PASEO_LOG_FILE_LEVEL",
"PASEO_LOG_FILE_PATH",
"PASEO_LOG_FILE_ROTATE_COUNT",
"PASEO_LOG_FILE_ROTATE_SIZE",
"PASEO_LOG_FORMAT",
"PASEO_LOG_LEVEL",
"PASEO_LOG_ROTATE_COUNT",
"PASEO_LOG_ROTATE_SIZE",
"PASEO_PASSWORD",
"PASEO_RELAY_ENABLED",
"PASEO_RELAY_ENDPOINT",
"PASEO_RELAY_PUBLIC_ENDPOINT",
"PASEO_RELAY_PUBLIC_USE_TLS",
"PASEO_RELAY_USE_TLS",
"PASEO_SERVICE_PROXY_ENABLED",
"PASEO_SERVICE_PROXY_LISTEN",
"PASEO_SERVICE_PROXY_PUBLIC_BASE_URL",
"PASEO_TRUSTED_PROXIES",
"PASEO_VOICE_LANGUAGE",
"PASEO_VOICE_LLM_PROVIDER",
"PASEO_VOICE_LOCAL_STT_MODEL",
"PASEO_VOICE_LOCAL_TTS_MODEL",
"PASEO_VOICE_LOCAL_TTS_SPEAKER_ID",
"PASEO_VOICE_LOCAL_TTS_SPEED",
"PASEO_VOICE_MODE_ENABLED",
"PASEO_VOICE_STT_PROVIDER",
"PASEO_VOICE_TTS_PROVIDER",
"PASEO_VOICE_TURN_DETECTION_PROVIDER",
"PASEO_WEB_UI_DIST_DIR",
"PASEO_WEB_UI_ENABLED",
"PORT",
"STT_CONFIDENCE_THRESHOLD",
"STT_MODEL",
"TTS_MODEL",
"TTS_VOICE",
] as const;
const CONFIG_CONTEXT_ENV_KEYS = [
"PASEO_NODE_ENV",
"PASEO_DESKTOP_MANAGED",
"OPENAI_API_KEY",
"OPENAI_BASE_URL",
"OPENAI_STT_API_KEY",
"OPENAI_TTS_API_KEY",
] as const;
export function configurationEnvironment(env: NodeJS.ProcessEnv): NodeJS.ProcessEnv {
return Object.fromEntries(
[...DAEMON_SETTING_ENV_KEYS, ...CONFIG_CONTEXT_ENV_KEYS].map((key) => [key, env[key]]),
);
}
export function daemonLaunchEnvironment(input: {
env: NodeJS.ProcessEnv;
home: string;
mode: "managed" | "deployment";
desktopManaged?: boolean;
}): NodeJS.ProcessEnv {
const env = { ...input.env };
if (input.mode === "managed") {
for (const key of DAEMON_SETTING_ENV_KEYS) delete env[key];
}
delete env.PASEO_HOST;
delete env.PASEO_DESKTOP_MANAGED;
env.PASEO_HOME = input.home;
if (input.desktopManaged) env.PASEO_DESKTOP_MANAGED = "1";
return env;
}
+2 -1
View File
@@ -1,3 +1,4 @@
import { configurationEnvironment } from "./config-environment.js";
import { existsSync } from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
@@ -554,7 +555,7 @@ export function resolveConfigFromPersisted(
options?: ResolveConfigFromPersistedOptions,
): PaseoDaemonConfig {
const resolvedOptions = options ?? {};
const env = resolvedOptions.env ?? process.env;
const env = configurationEnvironment(resolvedOptions.env ?? process.env);
const cli = resolvedOptions.cli;
const relayEnabledFallback =
resolvedOptions.relayEnabledFallback ?? persisted.daemon?.relay?.enabled === undefined;
@@ -0,0 +1,305 @@
import { spawn } from "node:child_process";
import { readFile } from "node:fs/promises";
import path from "node:path";
import { setTimeout as delay } from "node:timers/promises";
import {
getPidLockInfo,
isPidRunning,
isSamePidLock,
releasePidLock,
type PidLockInfo,
} from "./pid-lock.js";
import { daemonLaunchEnvironment } from "./config-environment.js";
import { readPersistedConfig } from "./persisted-config.js";
import treeKill from "tree-kill";
const killTree = (pid: number, signal: string): Promise<void> =>
new Promise((resolve, reject) =>
treeKill(pid, signal, (error) => {
if (error) {
reject(error);
return;
}
resolve();
}),
);
export { daemonLaunchEnvironment } from "./config-environment.js";
export {
isSamePidLock as isSameDaemonInstance,
type PidLockInfo as DaemonInstance,
} from "./pid-lock.js";
export class DaemonInstanceError extends Error {
constructor(
public readonly code: string,
message: string,
) {
super(message);
}
}
export async function readDaemonInstance(home: string): Promise<PidLockInfo | null> {
const lock = await getPidLockInfo(home);
return lock && isPidRunning(lock.pid) ? lock : null;
}
export function daemonLogPath(home: string): string {
try {
return path.resolve(home, readPersistedConfig(home).log?.file?.path ?? "daemon.log");
} catch {
return path.join(home, "daemon.log");
}
}
export async function waitForDaemonReady(
home: string,
options: {
timeoutMs?: number;
instance?: PidLockInfo;
signal?: AbortSignal;
} = {},
): Promise<PidLockInfo & { listen: string }> {
const deadline = Date.now() + (options.timeoutMs ?? 600_000);
while (true) {
options.signal?.throwIfAborted();
const instance = await readDaemonInstance(home);
if (!instance)
throw new DaemonInstanceError(
"DAEMON_NOT_RUNNING",
`Daemon is not running for ${home}. Start with: paseo daemon start --home ${JSON.stringify(home)}`,
);
if (options.instance && !isSamePidLock(instance, options.instance)) {
throw new DaemonInstanceError(
"DAEMON_REPLACED",
`Supervisor changed for ${home}; refusing to follow PID ${instance.pid}.`,
);
}
if (instance.listen) return { ...instance, listen: instance.listen };
if (Date.now() >= deadline) throw notReady(home, instance);
await delay(100, undefined, { signal: options.signal });
}
}
function notReady(home: string, instance: PidLockInfo): DaemonInstanceError {
return new DaemonInstanceError(
"DAEMON_NOT_READY",
`Daemon PID ${instance.pid} remains running but is not ready for ${home}.\nLogs: ${daemonLogPath(home)}\nStatus: paseo daemon status --home ${JSON.stringify(home)}\nStop: paseo daemon stop --home ${JSON.stringify(home)}`,
);
}
async function requestInstanceStop(
home: string,
instance: PidLockInfo,
options: {
force?: boolean;
requestShutdown?: (instance: PidLockInfo & { listen: string }) => Promise<void>;
},
) {
let forced = false;
let usedLifecycleRpc = false;
if (process.platform === "win32") {
if (instance.listen && options.requestShutdown) {
try {
await options.requestShutdown({ ...instance, listen: instance.listen });
usedLifecycleRpc = true;
} catch (error) {
if (!options.force) throw error;
const current = await getPidLockInfo(home);
if (current && !isSamePidLock(instance, current))
throw new DaemonInstanceError(
"DAEMON_REPLACED",
`Supervisor changed for ${home}; refusing forced cleanup.`,
);
await killTree(instance.pid, "SIGKILL");
forced = true;
}
} else if (!options.force) {
throw new DaemonInstanceError(
"STOP_NO_GRACEFUL_CHANNEL",
`PID ${instance.pid} for ${home} has no graceful shutdown channel. Use --force explicitly to terminate it.`,
);
} else {
await killTree(instance.pid, "SIGKILL");
forced = true;
}
} else {
try {
process.kill(instance.pid, "SIGTERM");
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error;
}
}
return { forced, usedLifecycleRpc };
}
export async function stopDaemonInstance(
home: string,
options: {
instance?: PidLockInfo;
force?: boolean;
timeoutMs?: number;
killTimeoutMs?: number;
requestShutdown?: (instance: PidLockInfo & { listen: string }) => Promise<void>;
} = {},
): Promise<{
action: "stopped" | "not_running";
pid: number | null;
forced: boolean;
usedLifecycleRpc: boolean;
}> {
const { timeoutMs = 15_000, killTimeoutMs = 3_000 } = options;
const deadline = Date.now() + timeoutMs;
const instance = await getPidLockInfo(home);
if (options.instance && instance && !isSamePidLock(instance, options.instance)) {
throw new DaemonInstanceError(
"DAEMON_REPLACED",
`Supervisor changed for ${home}; refusing to stop PID ${instance.pid}.`,
);
}
if (!instance || !isPidRunning(instance.pid)) {
if (instance)
await releasePidLock(home, { ownerPid: instance.pid, startedAt: instance.startedAt });
return {
action: "not_running",
pid: instance?.pid ?? null,
forced: false,
usedLifecycleRpc: false,
};
}
if (instance.pid <= 1 || instance.pid === process.pid)
throw new Error("Refusing to stop invalid supervisor PID");
let { forced, usedLifecycleRpc } = await requestInstanceStop(home, instance, options);
const waitForExit = async (waitMs: number) => {
const exitDeadline = Date.now() + waitMs;
while (isPidRunning(instance.pid)) {
const current = await getPidLockInfo(home);
if (current && !isSamePidLock(instance, current))
throw new DaemonInstanceError(
"DAEMON_REPLACED",
`Supervisor changed for ${home}; stop of PID ${instance.pid} was not confirmed.`,
);
if (Date.now() >= exitDeadline) return false;
await delay(100);
}
return true;
};
let stopped = await waitForExit(forced ? killTimeoutMs : Math.max(0, deadline - Date.now()));
if (!stopped && options.force && !forced) {
await killTree(instance.pid, "SIGKILL");
forced = true;
stopped = await waitForExit(killTimeoutMs);
}
if (!stopped)
throw new DaemonInstanceError(
"STOP_NOT_CONFIRMED",
`Timed out waiting for supervisor PID ${instance.pid} in ${home} to exit${options.force ? "" : "; use --force to permit forced cleanup"}.`,
);
await releasePidLock(home, { ownerPid: instance.pid, startedAt: instance.startedAt });
return { action: "stopped", pid: instance.pid, forced, usedLifecycleRpc };
}
export async function startDaemonInstance(input: {
home: string;
command: string;
args: string[];
env: NodeJS.ProcessEnv;
mode: "managed" | "deployment";
desktopManaged?: boolean;
foreground?: boolean;
timeoutMs?: number;
signal?: AbortSignal;
onAcquired?: (instance: PidLockInfo) => void;
onReady?: (instance: PidLockInfo & { listen: string }) => void;
}): Promise<{ instance: PidLockInfo; spawned: boolean; exitCode?: number }> {
input.signal?.throwIfAborted();
const existing = await readDaemonInstance(input.home);
if (existing) return { instance: existing, spawned: false };
const {
foreground = false,
timeoutMs = 600_000,
signal,
onAcquired = () => {},
onReady = () => {},
} = input;
const child = spawn(input.command, input.args, {
env: daemonLaunchEnvironment(input),
detached: !foreground,
stdio: foreground ? "inherit" : "ignore",
windowsHide: true,
});
let exit: { code: number; error?: Error } | undefined;
const exited = new Promise<number>((resolve) => {
child.once("error", (error) => {
exit = { code: 1, error };
resolve(1);
});
child.once("exit", (code) => {
exit = { code: code ?? 1 };
resolve(code ?? 1);
});
});
const cancel = () => {
if (child.pid && !exit) {
child.ref();
child.kill("SIGTERM");
}
};
signal?.addEventListener("abort", cancel, { once: true });
if (!foreground) child.unref();
const deadline = foreground ? Infinity : Date.now() + timeoutMs;
let acquired: PidLockInfo | undefined;
async function waitUntilReady() {
while (true) {
signal?.throwIfAborted();
const instance = await readDaemonInstance(input.home);
if (instance && instance.pid !== child.pid) return { instance, spawned: false };
if (instance && !acquired) {
acquired = instance;
onAcquired(instance);
}
if (exit) {
const logPath = daemonLogPath(input.home);
const log = await readFile(logPath, "utf8").catch(() => "");
throw new DaemonInstanceError(
"DAEMON_START_FAILED",
`Daemon failed to start (${exit.error?.message ?? `exit ${exit.code}`}). Logs: ${logPath}\n${log.split("\n").slice(-30).join("\n")}`,
);
}
if (instance?.listen) {
const ready = { ...instance, listen: instance.listen };
onReady(ready);
return {
instance: ready,
spawned: true,
...(foreground ? { exitCode: await exited } : {}),
};
}
if (Date.now() >= deadline) {
if (acquired) throw notReady(input.home, acquired);
throw new DaemonInstanceError(
"DAEMON_NOT_READY",
`Supervisor PID ${child.pid} remains running but has not published its lock for ${input.home}. Logs: ${daemonLogPath(input.home)}. Check paseo daemon status --home ${JSON.stringify(input.home)}. Stop with paseo daemon stop --home ${JSON.stringify(input.home)} once its lock is published, or signal this PID.`,
);
}
await delay(100, undefined, { signal });
}
}
try {
return await waitUntilReady();
} catch (error) {
if (!(error instanceof DaemonInstanceError && error.code === "DAEMON_NOT_READY")) {
cancel();
const stopped = await Promise.race([
exited.then(() => true),
delay(15_000, false, { ref: false }),
]);
if (!stopped && child.pid) {
await killTree(child.pid, "SIGKILL");
await exited;
}
}
throw error;
} finally {
signal?.removeEventListener("abort", cancel);
}
}
+2 -5
View File
@@ -1,6 +1,6 @@
import { appendFileSync, mkdirSync } from "node:fs";
import path from "node:path";
import { createPaseoDaemon } from "./bootstrap.js";
import { createPaseoDaemon, formatListenTarget } from "./bootstrap.js";
import { loadConfig } from "./config.js";
import { resolvePaseoHome } from "./paseo-home.js";
import { createRootLogger } from "./logger.js";
@@ -323,10 +323,7 @@ async function main() {
try {
await daemon.start();
const listenTarget = daemon.getListenTarget();
const listen =
listenTarget?.type === "tcp"
? `${listenTarget.host}:${listenTarget.port}`
: listenTarget?.path;
const listen = formatListenTarget(listenTarget);
if (!listen) {
throw new Error("Daemon did not expose a listen target after startup");
}
+20
View File
@@ -96,3 +96,23 @@ export type {
AgentStreamEventPayload,
AgentStreamMessage,
} from "@getpaseo/protocol/messages";
export {
readDaemonInstance,
startDaemonInstance,
stopDaemonInstance,
waitForDaemonReady,
daemonLogPath,
daemonLaunchEnvironment,
isSameDaemonInstance,
DaemonInstanceError,
type DaemonInstance,
} from "./daemon-instance.js";
export {
readPersistedConfig,
getPersistedConfigValue,
editPersistedConfig,
} from "./persisted-config.js";
export { resolveConfigFromPersisted } from "./config.js";
export { ensurePrivateDirectory } from "./private-files.js";
-2
View File
@@ -1,6 +1,5 @@
import os from "node:os";
import path from "node:path";
import { ensurePrivateDirectory } from "./private-files.js";
function expandHomeDir(input: string): string {
if (input.startsWith("~/")) {
@@ -15,6 +14,5 @@ function expandHomeDir(input: string): string {
export function resolvePaseoHome(env: NodeJS.ProcessEnv = process.env): string {
const raw = env.PASEO_HOME ?? "~/.paseo";
const resolved = path.resolve(expandHomeDir(raw));
ensurePrivateDirectory(resolved);
return resolved;
}
@@ -0,0 +1,45 @@
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
import { existsSync } from "node:fs";
import { tmpdir } from "node:os";
import path from "node:path";
import { expect, test } from "vitest";
import {
editPersistedConfig,
getPersistedConfigValue,
readPersistedConfig,
} from "./persisted-config.js";
test("configuration edits validate before writing and preserve unrelated settings", async () => {
const root = await mkdtemp(path.join(tmpdir(), "paseo-config-edit-"));
const home = path.join(root, "home");
try {
expect(readPersistedConfig(home)).toEqual({});
expect(existsSync(home)).toBe(false);
for (const field of [
"unknown",
"daemon.missing",
"daemon.__proto__.listen",
"daemon.auth.password",
]) {
expect(() => editPersistedConfig(home, field, { value: "value" })).toThrow();
expect(existsSync(home)).toBe(false);
}
editPersistedConfig(home, "daemon.listen", { value: "127.0.0.1:12345" });
editPersistedConfig(home, "features.webUi.enabled", { value: true });
const before = await readFile(path.join(home, "config.json"), "utf8");
expect(() => editPersistedConfig(home, "features.webUi.enabled", { value: "true" })).toThrow();
expect(() =>
editPersistedConfig(home, "daemon", { value: { auth: { password: "plaintext" } } }),
).toThrow(/set-password/);
expect(await readFile(path.join(home, "config.json"), "utf8")).toBe(before);
editPersistedConfig(home, "features.webUi.enabled", { unset: true });
const edited = readPersistedConfig(home);
expect(getPersistedConfigValue(edited, "features.webUi.enabled")).toBeUndefined();
expect(edited.daemon?.listen).toBe("127.0.0.1:12345");
await writeFile(path.join(home, "config.json"), "invalid json");
expect(() => editPersistedConfig(home, "daemon.listen", { value: "12345" })).toThrow();
expect(await readFile(path.join(home, "config.json"), "utf8")).toBe("invalid json");
} finally {
await rm(root, { recursive: true, force: true });
}
});
@@ -463,6 +463,84 @@ export function loadPersistedConfig(paseoHome: string, logger?: LoggerLike): Per
return result.data as PersistedConfig;
}
/** Observe the file without initializing a home, identity, or default configuration. */
export function readPersistedConfig(
paseoHome: string,
options: { defaultsIfMissing?: boolean } = {},
): PersistedConfig {
let raw: string;
try {
raw = readFileSync(getConfigPath(paseoHome), "utf8");
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT")
return options.defaultsIfMissing ? structuredClone(DEFAULT_PERSISTED_CONFIG) : {};
throw error;
}
return PersistedConfigSchema.parse(stripRemovedConfigFields(JSON.parse(raw))) as PersistedConfig;
}
function configPathParts(field: string): string[] {
const parts = field.split(".");
let schema: z.ZodType = PersistedConfigSchema;
for (const part of parts) {
if (!part || ["__proto__", "prototype", "constructor"].includes(part)) {
throw new Error(`Invalid configuration path: ${field}`);
}
while (
schema instanceof z.ZodOptional ||
schema instanceof z.ZodDefault ||
schema instanceof z.ZodPipe
)
schema = (schema instanceof z.ZodPipe ? schema.in : schema.unwrap()) as z.ZodType;
if (!(schema instanceof z.ZodObject) || !Object.hasOwn(schema.shape, part)) {
throw new Error(
`Unknown configuration path: ${field}. Replace its containing object with JSON for dynamic keys.`,
);
}
schema = schema.shape[part];
}
return parts;
}
export function getPersistedConfigValue(config: PersistedConfig, field: string): unknown {
return configPathParts(field).reduce<unknown>(
(value, part) =>
value && typeof value === "object" ? (value as Record<string, unknown>)[part] : undefined,
config,
);
}
export function editPersistedConfig(
paseoHome: string,
field: string,
edit: { value: unknown } | { unset: true },
): PersistedConfig {
const parts = configPathParts(field);
if (field === "daemon.auth" || field.startsWith("daemon.auth.")) {
throw new Error("Use daemon set-password to change the daemon password.");
}
const config = readPersistedConfig(paseoHome, { defaultsIfMissing: true });
let object = config as Record<string, unknown>;
for (const part of parts.slice(0, -1)) {
object[part] ??= {};
object = object[part] as Record<string, unknown>;
}
const key = parts.at(-1)!;
if (field === "daemon") {
const previousAuth = config.daemon?.auth;
const nextAuth =
"value" in edit && edit.value && typeof edit.value === "object"
? (edit.value as Record<string, unknown>).auth
: undefined;
if (JSON.stringify(previousAuth) !== JSON.stringify(nextAuth))
throw new Error("Use daemon set-password to change the daemon password.");
}
if ("unset" in edit) delete object[key];
else object[key] = edit.value;
savePersistedConfig(paseoHome, config);
return config;
}
export function savePersistedConfig(
paseoHome: string,
config: PersistedConfig,
+12 -14
View File
@@ -92,7 +92,7 @@ describe("pid-lock ownership", () => {
}
});
test("reclaims a stale desktop heartbeat lock after desktop confirms the daemon is unreachable", async () => {
test("preserves a stale live desktop heartbeat lock", async () => {
const paseoHome = await mkdtemp(join(tmpdir(), "paseo-pid-lock-stale-desktop-heartbeat-"));
const replacementOwnerPid = process.pid + 10_000;
@@ -113,14 +113,13 @@ describe("pid-lock ownership", () => {
const staleTime = new Date(Date.now() - 10 * 60_000);
await utimes(pidPath, staleTime, staleTime);
await acquirePidLock(paseoHome, null, {
ownerPid: replacementOwnerPid,
reclaimStaleDesktopLock: true,
});
await expect(
acquirePidLock(paseoHome, null, { ownerPid: replacementOwnerPid }),
).rejects.toThrow("Another Paseo daemon is already running");
const lock = await getPidLockInfo(paseoHome);
expect(lock?.pid).toBe(replacementOwnerPid);
expect(lock?.listen).toBeNull();
expect(lock?.pid).toBe(process.pid);
expect(lock?.listen).toBe("127.0.0.1:6767");
} finally {
await rm(paseoHome, { recursive: true, force: true });
}
@@ -156,7 +155,7 @@ describe("pid-lock ownership", () => {
}
});
test("reclaims a stale legacy desktop lock after desktop confirms the daemon is unreachable", async () => {
test("preserves a stale live legacy desktop lock", async () => {
const paseoHome = await mkdtemp(join(tmpdir(), "paseo-pid-lock-legacy-desktop-"));
const replacementOwnerPid = process.pid + 10_000;
const pidPath = join(paseoHome, "paseo.pid");
@@ -176,14 +175,13 @@ describe("pid-lock ownership", () => {
const staleTime = new Date(Date.now() - 10 * 60_000);
await utimes(pidPath, staleTime, staleTime);
await acquirePidLock(paseoHome, null, {
ownerPid: replacementOwnerPid,
reclaimStaleDesktopLock: true,
});
await expect(
acquirePidLock(paseoHome, null, { ownerPid: replacementOwnerPid }),
).rejects.toThrow("Another Paseo daemon is already running");
const lock = await getPidLockInfo(paseoHome);
expect(lock?.pid).toBe(replacementOwnerPid);
expect(lock?.heartbeat).toBe(true);
expect(lock?.pid).toBe(process.pid);
expect(lock?.heartbeat).toBeUndefined();
} finally {
await rm(paseoHome, { recursive: true, force: true });
}
+39 -57
View File
@@ -1,12 +1,12 @@
import { open, readFile, stat, unlink, mkdir, utimes } from "node:fs/promises";
import { open, readFile, unlink, utimes } from "node:fs/promises";
import type { FileHandle } from "node:fs/promises";
import { existsSync } from "node:fs";
import { ensurePrivateDirectory } from "./private-files.js";
import { join } from "node:path";
import { hostname } from "node:os";
import { z } from "zod";
export const pidLockInfoSchema = z.object({
pid: z.number(),
pid: z.number().int().positive(),
startedAt: z.string(),
hostname: z.string(),
uid: z.number(),
@@ -36,18 +36,16 @@ export class PidLockError extends Error {
}
}
// Stale recovery is for abandoned locks, so keep this well above ordinary event-loop stalls.
const PID_LOCK_STALE_MS = 5 * 60_000;
const PID_LOCK_HEARTBEAT_INTERVAL_MS = 30_000;
const PID_LOCK_READ_RETRY_ATTEMPTS = 10;
const PID_LOCK_READ_RETRY_DELAY_MS = 50;
function isPidRunning(pid: number): boolean {
export function isPidRunning(pid: number): boolean {
try {
process.kill(pid, 0);
return true;
} catch {
return false;
} catch (error) {
return isErrnoException(error) && error.code === "EPERM";
}
}
@@ -55,27 +53,29 @@ function getPidFilePath(paseoHome: string): string {
return join(paseoHome, "paseo.pid");
}
async function isPidLockFresh(pidPath: string): Promise<boolean> {
try {
const lockStat = await stat(pidPath);
return lockStat.mtimeMs >= Date.now() - PID_LOCK_STALE_MS;
} catch {
return false;
}
}
async function touchPidLockFile(pidPath: string): Promise<void> {
const now = new Date();
await utimes(pidPath, now, now);
}
async function readPidLock(pidPath: string): Promise<PidLockInfo | null> {
try {
const content = await readFile(pidPath, "utf-8");
return parsePidLockInfo(JSON.parse(content));
} catch {
return null;
let lastError: unknown;
for (let attempt = 0; attempt < PID_LOCK_READ_RETRY_ATTEMPTS; attempt++) {
try {
const content = await readFile(pidPath, "utf-8");
const lock = parsePidLockInfo(JSON.parse(content));
if (lock) return lock;
lastError = new Error("Invalid lock shape");
} catch (error) {
if (isErrnoException(error) && error.code === "ENOENT") return null;
lastError = error;
}
await new Promise((resolve) => setTimeout(resolve, PID_LOCK_READ_RETRY_DELAY_MS));
}
throw Object.assign(
new PidLockError(`Cannot read daemon state at ${pidPath}: ${String(lastError)}`),
{ code: "DAEMON_STATE_READ_FAILED" },
);
}
function resolveOwnerPid(ownerPid?: number): number {
@@ -87,19 +87,9 @@ function resolveOwnerPid(ownerPid?: number): number {
interface AcquirePidLockOptions {
ownerPid?: number;
reclaimStaleDesktopLock?: boolean;
}
function canReclaimLiveLock(
lock: PidLockInfo,
options: AcquirePidLockOptions | undefined,
): boolean {
// COMPAT(pidLockHeartbeat): v0.1.108 desktop startup has already confirmed the old daemon is
// unreachable before it launches the supervisor. Remove after 2027-01-15.
return options?.reclaimStaleDesktopLock === true && lock.desktopManaged === true;
}
function isSamePidLock(left: PidLockInfo, right: PidLockInfo): boolean {
export function isSamePidLock(left: PidLockInfo, right: PidLockInfo): boolean {
return left.pid === right.pid && left.startedAt === right.startedAt;
}
@@ -114,7 +104,6 @@ async function clearExistingPidLock(
pidPath: string,
existingLock: PidLockInfo,
lockOwnerPid: number,
options: AcquirePidLockOptions | undefined,
): Promise<"already_owned" | "cleared"> {
const lockOwnerRunning = isPidRunning(existingLock.pid);
if (existingLock.pid === lockOwnerPid && lockOwnerRunning) {
@@ -122,21 +111,14 @@ async function clearExistingPidLock(
return "already_owned";
}
if (lockOwnerRunning) {
const reclaimable = canReclaimLiveLock(existingLock, options);
if (!reclaimable || (await isPidLockFresh(pidPath))) {
throw createLockHeldError(existingLock);
}
// Re-read immediately before unlinking so a heartbeat at the stale boundary wins.
const confirmedLock = await readPidLock(pidPath);
if (
!confirmedLock ||
!isSamePidLock(existingLock, confirmedLock) ||
(await isPidLockFresh(pidPath))
) {
throw new PidLockError("PID lock changed while checking whether it was abandoned");
}
if (lockOwnerRunning) throw createLockHeldError(existingLock);
const confirmedLock = await readPidLock(pidPath);
if (
!confirmedLock ||
!isSamePidLock(existingLock, confirmedLock) ||
isPidRunning(confirmedLock.pid)
) {
throw new PidLockError("PID lock changed while checking whether it was abandoned");
}
await unlink(pidPath).catch(() => {});
@@ -173,10 +155,7 @@ export async function acquirePidLock(
): Promise<void> {
const pidPath = getPidFilePath(paseoHome);
// Ensure paseoHome directory exists
if (!existsSync(paseoHome)) {
await mkdir(paseoHome, { recursive: true });
}
ensurePrivateDirectory(paseoHome);
// Try to read existing lock
const existingLock = await readPidLock(pidPath);
@@ -184,7 +163,7 @@ export async function acquirePidLock(
// Check if existing lock is stale
const lockOwnerPid = resolveOwnerPid(options?.ownerPid);
if (existingLock) {
const result = await clearExistingPidLock(pidPath, existingLock, lockOwnerPid, options);
const result = await clearExistingPidLock(pidPath, existingLock, lockOwnerPid);
if (result === "already_owned") {
return;
}
@@ -298,7 +277,7 @@ export function startPidLockHeartbeat(
export async function updatePidLock(
paseoHome: string,
patch: { listen: string },
patch: { listen: string | null },
options?: { ownerPid?: number },
): Promise<void> {
const pidPath = getPidFilePath(paseoHome);
@@ -329,7 +308,7 @@ export async function updatePidLock(
export async function releasePidLock(
paseoHome: string,
options?: { ownerPid?: number },
options?: { ownerPid?: number; startedAt?: string },
): Promise<void> {
const pidPath = getPidFilePath(paseoHome);
const lockOwnerPid = resolveOwnerPid(options?.ownerPid);
@@ -337,7 +316,10 @@ export async function releasePidLock(
// Only remove if it's our lock
const content = await readFile(pidPath, "utf-8");
const lock = parsePidLockInfo(JSON.parse(content));
if (lock?.pid === lockOwnerPid) {
if (
lock?.pid === lockOwnerPid &&
(options?.startedAt === undefined || lock.startedAt === options.startedAt)
) {
await unlink(pidPath);
}
} catch {