From 68fc16b85507b9fe4ce7830e0293c6d95999afd2 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 6 Nov 2026 20:20:18 +0700 Subject: [PATCH] fix(terminal): run terminal workers without process identity queries (#8818) * fix(terminal): run terminals when process identity cannot be read On some Windows machines Windows PowerShell cannot run Get-CimInstance, or cannot be started at all. The terminal worker refused to start without its own birth time and command line, so those machines had no terminals. Process identity only authorizes signals. The worker now starts anyway and records its identity as unavailable, and programs without a birth time. Cleanup never signals what it cannot identify. It asks such a worker to shut down over the socket the secret authenticates, and reports survivors as unconfirmed. So records of crashed workers do not block launches, the process table falls back to process.kill(pid, 0) when the query fails. When a live pid cannot be identified, a refused connection on the worker's socket marks it as another process, because a worker listens until it exits. * fix(terminal): ask unidentified workers to shut down on every cleanup Retries of an unconfirmed cleanup and startup cleanup run without a grace period, which skipped the shutdown request. For a worker whose identity is unavailable that request is the only way to stop it, so it is now always sent. A gone program recorded without a birth time is no longer walked through the process table: the table was unreadable when it started, and the walk only kept the cleanup unconfirmed. Also test the socket probe on every platform, restore a test of the daemon starting when the worker cannot, and keep one pid liveness helper. * test(terminal): fail the worker launch the same way on every platform Windows opens a directory sitting at the launcher's log path, so the worker started there and the #8810 test never saw a failed launch. An invalid protocol override makes the worker exit before it writes its record on every platform. * fix(terminal): identify workers through their recorded sockets * fix(terminal): authenticate legacy sockets before forced cleanup * test(terminal): exercise closing records without process queries * test(terminal): identify stale lifecycle records by their sockets * refactor(terminal): drop the unreleased unavailable-identity record shape No release wrote identity: { unavailable }; it existed only between commits of this change. Records keep reading the identity workers up to 0.15.1 wrote, and nothing consults it. --- docs/terminal-worker.md | 13 +- .../terminal-worker-lifecycle.e2e.test.ts | 33 +- .../terminal/terminal-worker-client.test.ts | 47 ++- .../src/terminal/terminal-worker-client.ts | 34 +- .../terminal-worker-ownership.test.ts | 46 +-- .../src/terminal/terminal-worker-ownership.ts | 66 ++-- .../src/terminal/terminal-worker-process.ts | 33 +- .../src/terminal/terminal-worker-processes.ts | 55 +--- .../terminal/terminal-worker-reaper.test.ts | 147 ++++++++- .../src/terminal/terminal-worker-reaper.ts | 50 +-- .../src/terminal/terminal-worker-record.ts | 17 +- .../terminal-worker-socket-path.test.ts | 18 +- .../terminal/terminal-worker-socket-path.ts | 3 +- .../terminal/worker-terminal-manager.test.ts | 298 ++++++++++++++---- .../src/terminal/worker-terminal-manager.ts | 45 ++- 15 files changed, 648 insertions(+), 257 deletions(-) diff --git a/docs/terminal-worker.md b/docs/terminal-worker.md index be9321fb4f..77aa9325b7 100644 --- a/docs/terminal-worker.md +++ b/docs/terminal-worker.md @@ -21,11 +21,14 @@ Which triggers keep terminals and which close them is owned by the stop reason ( ## Record, socket and secret -The worker writes `$PASEO_HOME/terminal-worker/.json` (owner-only) with its identity (birth time + command line carrying `--paseo-terminal-worker `), protocol version, socket path, a random secret, the copy it runs from, and each terminal program's pid and birth time. +The worker writes `$PASEO_HOME/terminal-worker/.json` (owner-only) with its pid, protocol version, socket path, random secret, runtime copy and terminal programs. One daemon holds the home's pid lock. Its records and sockets belong to that home. -- The first frame on the socket must be `hello` with the secret; anything else closes the connection. Anyone who connects can type into every shell, so the record is `0600`, the unix socket `0600` in a `0700` directory. -- Unix socket paths have a byte budget (103 on macOS, 107 on Linux). Dev homes under worktrees blow it, so `terminal-worker-socket-path.ts` tries `$PASEO_HOME/terminal-worker/sock/`, then `$TMPDIR/paseo-/`, then `/tmp/paseo-/`, and accepts an existing directory only if it is a real `0700` directory owned by you. It never chmods a shared root. -- Cleanup signals a process only when its birth time matches the record. Birth times come from `ps -o lstart` with `LC_ALL=C` (it follows `LC_TIME` otherwise) and from `Win32_Process.CreationDate`. A missing birth time is "unknown", never matched by command text: a shell that ran `exec` keeps its pid and birth time but not its argv. +- The first frame must be `hello` with the secret; anything else closes the connection. Anyone authenticated can type into every shell, so the record and Unix socket are `0600` inside a `0700` directory. +- New socket names include the record id. A stale record whose pid belongs to a newer worker cannot reach that newer worker. Existing records keep their recorded paths. Legacy paths use only the home and pid, so force-killing through one also requires successful authentication. A legacy socket that cannot authenticate stays unconfirmed. +- A missing pid or refused socket retires the record without signaling the worker. An accepting socket identifies the live worker. Cleanup asks it to shut down, then probes again before force-killing it after the grace period (`taskkill /T` on Windows). An inconclusive probe counts toward the worker limit and is retried on the next start or stop without signaling it. +- Workers never query their own birth time or command line. Readers accept legacy identity fields but ignore them. Daemons through 0.15.1 report new records unreadable and cannot adopt them. +- Windows terminal programs have no recorded birth time and terminal lifecycle operations never launch PowerShell. Linux and macOS retain program birth-time checks through `ps` with `LC_ALL=C`; Linux also walks sessions to find surviving jobs. +- Unix socket paths have a byte budget (103 on macOS, 107 on Linux). Long dev homes use an owner-only fallback under `$TMPDIR/paseo-/` or `/tmp/paseo-/`. An existing directory must be a real `0700` directory owned by you; a shared root is never chmodded. - A kill cleanup that cannot confirm every process gone writes `closing-.json` beside the worker record. The worker rewrites its own record whenever a terminal starts or exits, so the kill intent has to live in a file it never writes. A closing record is never adopted; the next stop or start retries it. - An unreadable record of either kind makes `daemon status` and a kill restart fail with its path: a kill cleanup that met it could not check what it names, and leaves nothing else behind. A read that fails with `EPERM`, `EBUSY` or `EACCES` is retried for 250ms first: on Windows, opening a record while the worker renames its new copy over it fails that way for a moment. @@ -40,7 +43,7 @@ The worker writes `$PASEO_HOME/terminal-worker/.json` (owner-only) with its | two or more older | shuts it down; its terminals close (logged) | | newer, or a version it cannot read | leaves it running, undriven; `list_terminals` and `create_terminal` answer `rpc_error` `terminal_worker_incompatible`, and `paseo daemon status` prints why | -Never more than two workers run. A new worker launches only when none is current, none is incompatible and fewer than two run. The count comes from the records, so an unreachable or unidentified worker counts, and so does one a cleanup could not confirm gone. When two run, the daemon starts with no current worker, and the first terminal created after one of them exits launches it. +Never more than two workers run. A new worker launches only when none is current, none is incompatible and fewer than two run. The count comes from the records, so an unreachable worker counts, as does one whose socket probe is inconclusive, and one a cleanup could not confirm gone. When two run, the daemon starts with no current worker, and the first terminal created after one of them exits launches it. **The frozen envelope never changes across versions:** the frame format (4-byte length + `v8.serialize`), `hello`/`helloAck`, the `listTerminals` info fields and `shutdown`. That is what lets any daemon close any worker: a kill stop sends `shutdown` to a newer worker it cannot otherwise drive. Protocol 5 hard cuts unreleased protocols 1 through 4, including the acknowledgement-based replay contract: those workers are obsolete and close. For future versions, keep a previous-version adapter so the draining worker can still be driven, and drop the adapter for the version before that. diff --git a/packages/server/src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts b/packages/server/src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts index d606b4d4d9..a9789c906e 100644 --- a/packages/server/src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/terminal-worker-lifecycle.e2e.test.ts @@ -18,6 +18,7 @@ import { DaemonClient } from "../test-utils/daemon-client.js"; import { resolveDaemonVersion } from "../daemon-version.js"; import { readDaemonInstance, stopDaemonInstance } from "../daemon-instance.js"; import { currentBootId } from "../pid-lock.js"; +import { resolveTerminalWorkerSocketPath } from "../../terminal/terminal-worker-socket-path.js"; import { connectTerminalWorker } from "../../terminal/terminal-worker-client.js"; import { createTerminalWorkerProcessInspector } from "../../terminal/terminal-worker-processes.js"; import { reapTerminalWorkers } from "../../terminal/terminal-worker-reaper.js"; @@ -316,14 +317,6 @@ function spawnUnrelated(): ChildProcess { return child; } -async function birthOf(pid: number): Promise { - const inspection = await inspector.inspect(pid); - if (inspection.status !== "alive" || !inspection.snapshot.startedAt) { - throw new Error(`no birth time for ${pid}`); - } - return inspection.snapshot.startedAt; -} - afterEach(async () => { for (const harness of harnesses.splice(0)) { const pid = harness.supervisor.pid; @@ -582,28 +575,26 @@ describe("terminals when the daemon stops", { timeout: 240_000 }, () => { const reused = spawnUnrelated(); const earlierBoot = spawnUnrelated(); const base = await currentWorkerRecord(harness.home); - // Same pid, another birth time: what a reused pid looks like, whatever its command line. + // A reused pid with its old record-specific socket gone must never be signaled. writeTerminalWorkerRecord(harness.home, { ...base, recordId: "stale-record", pid: reused.pid!, - identity: { startedAt: "Thu Jan 1 00:00:00 1970", commandLine: "node -e setInterval" }, + socketPath: resolveTerminalWorkerSocketPath({ + paseoHome: harness.home, + workerPid: reused.pid!, + recordId: "stale-record", + }), terminals: [{ id: "t-stale", pid: reused.pid!, startedAt: "Thu Jan 1 00:00:00 1970" }], }); - // The right birth time, but written before this boot. + // Records from an earlier boot are retired before checking any process. writeTerminalWorkerRecord(harness.home, { ...base, recordId: "earlier-boot-record", pid: earlierBoot.pid!, bootId: currentBootId() ? "another-boot" : null, createdAt: "2000-01-01T00:00:00.000Z", - identity: { - startedAt: await birthOf(earlierBoot.pid!), - commandLine: `node --paseo-terminal-worker earlier-boot-record`, - }, - terminals: [ - { id: "t-boot", pid: earlierBoot.pid!, startedAt: await birthOf(earlierBoot.pid!) }, - ], + terminals: [{ id: "t-boot", pid: earlierBoot.pid!, startedAt: null }], }); await stopDaemon(harness); @@ -616,7 +607,11 @@ describe("terminals when the daemon stops", { timeout: 240_000 }, () => { ...base, recordId: "stale-record", pid: reused.pid!, - identity: { startedAt: "Thu Jan 1 00:00:00 1970", commandLine: "node -e setInterval" }, + socketPath: resolveTerminalWorkerSocketPath({ + paseoHome: harness.home, + workerPid: reused.pid!, + recordId: "stale-record", + }), terminals: [{ id: "t-stale", pid: reused.pid!, startedAt: "Thu Jan 1 00:00:00 1970" }], }); await stopDaemonInstance(harness.home, { force: true }); diff --git a/packages/server/src/terminal/terminal-worker-client.test.ts b/packages/server/src/terminal/terminal-worker-client.test.ts index 01a4dd91d3..119d339b18 100644 --- a/packages/server/src/terminal/terminal-worker-client.test.ts +++ b/packages/server/src/terminal/terminal-worker-client.test.ts @@ -4,7 +4,13 @@ import { tmpdir } from "node:os"; import path from "node:path"; import { isPlatform } from "../test-utils/platform.js"; import { encodeFrame, FrameDecoder, TerminalWorkerFrameError } from "./terminal-worker-frames.js"; -import { connectTerminalWorker, TerminalWorkerConnectError } from "./terminal-worker-client.js"; +import { createConnection, createServer, type Server } from "node:net"; +import { randomUUID } from "node:crypto"; +import { + connectTerminalWorker, + probeTerminalWorkerSocket, + TerminalWorkerConnectError, +} from "./terminal-worker-client.js"; import { launchTerminalWorker } from "./terminal-worker-launch.js"; import { terminalWorkerRecordPath } from "./terminal-worker-record.js"; @@ -74,7 +80,6 @@ describe("terminal worker authentication", () => { }); it("bounds a stalled framed writer to one chunk and resumes in order", async () => { - const { createServer, createConnection } = await import("node:net"); const { attachFramedSocket } = await import("./terminal-worker-frames.js"); const server = createServer(); await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); @@ -115,3 +120,41 @@ it("bounds a stalled framed writer to one chunk and resumes in order", async () await new Promise((resolve) => server.close(() => resolve())); } }); + +// The probe decides whether a live pid named by an unidentifiable worker record is still the +// worker. A refused answer retires the record, so it must hold for a Windows named pipe too. +describe("terminal worker socket probe", () => { + function socketPath(): string { + if (isPlatform("win32")) return `\\\\.\\pipe\\paseo-tw-probe-${randomUUID()}`; + const directory = mkdtempSync(path.join(tmpdir(), "tw-probe-")); + temporaryDirs.push(directory); + return path.join(directory, "w.sock"); + } + + async function serve(at: string): Promise { + const server = createServer((socket) => socket.on("error", () => {})); + await new Promise((resolve) => server.listen(at, resolve)); + return server; + } + + it("answers accepting while a worker serves the socket", async () => { + const at = socketPath(); + const server = await serve(at); + try { + expect(await probeTerminalWorkerSocket(at)).toBe("accepting"); + } finally { + await new Promise((resolve) => server.close(resolve)); + } + }); + + it("answers refused for a socket nobody ever served", async () => { + expect(await probeTerminalWorkerSocket(socketPath())).toBe("refused"); + }); + + it("answers refused once the worker that served the socket closed it", async () => { + const at = socketPath(); + const server = await serve(at); + await new Promise((resolve) => server.close(resolve)); + expect(await probeTerminalWorkerSocket(at)).toBe("refused"); + }); +}); diff --git a/packages/server/src/terminal/terminal-worker-client.ts b/packages/server/src/terminal/terminal-worker-client.ts index 532f3075c6..ac48293879 100644 --- a/packages/server/src/terminal/terminal-worker-client.ts +++ b/packages/server/src/terminal/terminal-worker-client.ts @@ -1,4 +1,5 @@ -import { EventEmitter } from "node:events"; +import { EventEmitter, once } from "node:events"; +import { setTimeout as delay } from "node:timers/promises"; import { createConnection } from "node:net"; import { attachFramedSocket, type FramedSocket } from "./terminal-worker-frames.js"; import { @@ -139,3 +140,34 @@ export function connectTerminalWorker(input: { }); }); } + +export type TerminalWorkerSocketProbe = "accepting" | "refused" | "unknown"; + +const PROBE_TIMEOUT_MS = 1_000; +// What a connect to a socket nobody listens on fails with: the unix socket file is left behind +// (ECONNREFUSED) or gone, as a Windows pipe is once its server closes (ENOENT). +const NO_LISTENER_ERRORS = new Set(["ECONNREFUSED", "ENOENT"]); + +/** + * Whether anything accepts connections on a worker's socket. It never authenticates, so the + * daemon controlling the worker keeps control: the worker drops a connection that closes + * before its hello. + */ +export async function probeTerminalWorkerSocket( + socketPath: string, +): Promise { + const socket = createConnection(socketPath); + const timeout = new AbortController(); + try { + return await Promise.race([ + once(socket, "connect").then(() => "accepting" as const), + delay(PROBE_TIMEOUT_MS, "unknown" as const, { signal: timeout.signal }), + ]); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + return code && NO_LISTENER_ERRORS.has(code) ? "refused" : "unknown"; + } finally { + timeout.abort(); + socket.destroy(); + } +} diff --git a/packages/server/src/terminal/terminal-worker-ownership.test.ts b/packages/server/src/terminal/terminal-worker-ownership.test.ts index 7abbb9aa62..aefa364338 100644 --- a/packages/server/src/terminal/terminal-worker-ownership.test.ts +++ b/packages/server/src/terminal/terminal-worker-ownership.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "vitest"; import type { ManagedProcessInspection } from "../server/managed-processes/managed-processes.js"; import { classifyTerminalProgramProcess, - classifyTerminalWorkerProcess, + identifyTerminalWorker, } from "./terminal-worker-ownership.js"; const BIRTH = "Wed Oct 21 18:12:49 2026"; @@ -12,14 +12,6 @@ function alive(startedAt: string | null, commandLine: string | null): ManagedPro return { status: "alive", snapshot: { pid: 4242, startedAt, commandLine } }; } -const workerRecord = { - recordId: "record-1", - identity: { - startedAt: BIRTH, - commandLine: "node terminal-worker-process.js --paseo-terminal-worker record-1", - }, -}; - describe("terminal program ownership", () => { it("owns a program that ran exec: same pid and birth time, new command line", () => { expect( @@ -59,33 +51,15 @@ describe("terminal program ownership", () => { }); describe("terminal worker ownership", () => { - it("owns the worker whose birth time matches and whose command line carries its record id", () => { + it("retires a live unrelated pid when its socket is absent without inspecting it", async () => { expect( - classifyTerminalWorkerProcess(workerRecord, alive(BIRTH, workerRecord.identity.commandLine)), - ).toEqual({ status: "owned" }); - }); - - it("treats another worker reusing the pid in the same second as unrelated", () => { - expect( - classifyTerminalWorkerProcess( - workerRecord, - alive(BIRTH, "node terminal-worker-process.js --paseo-terminal-worker record-2"), - ), - ).toMatchObject({ status: "unrelated" }); - }); - - it("treats a worker pid reused with another birth time as unrelated", () => { - expect( - classifyTerminalWorkerProcess( - workerRecord, - alive(OTHER_BIRTH, workerRecord.identity.commandLine), - ), - ).toMatchObject({ status: "unrelated" }); - }); - - it("does not own a worker whose command line cannot be read", () => { - expect(classifyTerminalWorkerProcess(workerRecord, alive(BIRTH, null))).toMatchObject({ - status: "unknown", - }); + await identifyTerminalWorker({ + pid: process.pid, + socketPath: + process.platform === "win32" + ? "\\\\.\\pipe\\paseo-absent-ownership-test" + : "/tmp/paseo-absent-ownership-test.sock", + }), + ).toEqual({ status: "gone" }); }); }); diff --git a/packages/server/src/terminal/terminal-worker-ownership.ts b/packages/server/src/terminal/terminal-worker-ownership.ts index 9bd02b01cd..0f7033f1b3 100644 --- a/packages/server/src/terminal/terminal-worker-ownership.ts +++ b/packages/server/src/terminal/terminal-worker-ownership.ts @@ -1,5 +1,6 @@ import type { ManagedProcessInspection } from "../server/managed-processes/managed-processes.js"; -import { TERMINAL_WORKER_MARKER_ARG, type TerminalWorkerRecord } from "./terminal-worker-record.js"; +import { connectTerminalWorker, probeTerminalWorkerSocket } from "./terminal-worker-client.js"; +import { type TerminalWorkerRecord } from "./terminal-worker-record.js"; // Who a pid named in a terminal worker record belongs to, before anything signals it. // @@ -15,20 +16,49 @@ export type ProcessOwnership = | { status: "owned" } | { status: "unknown"; reason: string }; -export function classifyTerminalWorkerProcess( - record: Pick, - inspection: ManagedProcessInspection, -): ProcessOwnership { - const birth = classifyBirth(record.identity.startedAt, inspection); - if (birth.status !== "owned" || inspection.status !== "alive") return birth; - const commandLine = inspection.snapshot.commandLine; - if (!commandLine) { - return { status: "unknown", reason: "command line unavailable" }; +/** The home lock owns these records; the record-specific socket proves a live worker. */ +export async function identifyTerminalWorker( + record: Pick, +): Promise { + try { + process.kill(record.pid, 0); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ESRCH") return { status: "gone" }; } - if (!hasWorkerMarker(commandLine, record.recordId)) { - return { status: "unrelated", reason: "command line lacks the worker marker" }; + const socket = await probeTerminalWorkerSocket(record.socketPath); + if (socket === "refused") return { status: "gone" }; + if (socket === "accepting") return { status: "owned" }; + return { status: "unknown", reason: "worker socket probe inconclusive" }; +} + +/** Rechecks the live worker immediately before a forced termination. */ +export async function authorizeTerminalWorkerTermination( + record: TerminalWorkerRecord, +): Promise { + const ownership = await identifyTerminalWorker(record); + if (ownership.status !== "owned") return ownership; + // COMPAT(terminal-worker-legacy-socket): added after v0.15.1, remove after 2027-02-06. + // Legacy names contain only home + pid. A stale record can reach a newer worker, so an + // accepting legacy socket requires proof of its secret before cleanup may signal it. + if ( + record.socketPath.endsWith(`-${record.recordId}.sock`) || + record.socketPath.endsWith(`-${record.recordId}`) + ) + return ownership; + try { + const connection = await connectTerminalWorker({ + socketPath: record.socketPath, + secret: record.secret, + timeoutMs: 1000, + }); + const matches = connection.workerPid === record.pid; + connection.disconnect(); + return matches + ? ownership + : { status: "unknown", reason: "legacy socket authenticated another pid" }; + } catch { + return { status: "unknown", reason: "legacy socket ownership could not be authenticated" }; } - return { status: "owned" }; } export function classifyTerminalProgramProcess( @@ -57,13 +87,3 @@ function classifyBirth( } return { status: "owned" }; } - -function hasWorkerMarker(commandLine: string, recordId: string): boolean { - const tokens = commandLine.split(/\s+/); - for (let index = 0; index < tokens.length - 1; index += 1) { - if (tokens[index] === TERMINAL_WORKER_MARKER_ARG && tokens[index + 1] === recordId) { - return true; - } - } - return false; -} diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index 855c39da59..b9eb4698c0 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -403,9 +403,10 @@ function runWorker(argv: string[]): void { ]); } - // A terminal enters the record only with its program's birth time, so cleanup can always - // tell the program from a later process that reuses its pid. A program that already exited - // needs no entry; one whose birth time cannot be read is killed and the create fails. + // A terminal enters the record with its program's birth time, so cleanup can tell the + // program from a later process that reuses its pid. A program that already exited needs no + // entry. One whose birth time cannot be read is recorded without it: it still runs, and + // cleanup never signals it. async function recordTerminalProgram(session: LocalTerminalSession): Promise { if (session.getExitInfo()) return; // The manager drops a session as soon as its program exits, so only a live one has a pid. @@ -413,14 +414,17 @@ function runWorker(argv: string[]): void { if (pid === null) { throw new Error(`Terminal ${session.id} has no program pid`); } - const inspection = await inspector.inspect(pid); - if (inspection.status === "not-found" || session.getExitInfo()) { + const inspection = inspector.reportsBirthTimes ? await inspector.inspect(pid) : null; + if (inspection?.status === "not-found" || session.getExitInfo()) { return; } - if (inspection.status === "error" || !inspection.snapshot.startedAt) { + const startedAt = inspection?.status === "alive" ? inspection.snapshot.startedAt : null; + if (inspector.reportsBirthTimes && !startedAt) { const reason = - inspection.status === "error" ? errorMessage(inspection.error) : "no birth time reported"; - throw new Error(`Cannot identify terminal program ${pid}: ${reason}`); + inspection?.status === "error" ? errorMessage(inspection.error) : "no birth time reported"; + console.error( + `Terminal worker cannot read the birth time of terminal program ${pid}; cleanup will never signal it: ${reason}`, + ); } if (!record) { throw new Error("Terminal worker record is not written yet"); @@ -429,7 +433,7 @@ function runWorker(argv: string[]): void { ...record, terminals: [ ...record.terminals.filter((entry) => entry.id !== session.id), - { id: session.id, pid, startedAt: inspection.snapshot.startedAt }, + { id: session.id, pid, startedAt }, ], }; await persistRecord(); @@ -694,6 +698,7 @@ function runWorker(argv: string[]): void { const socketPath = resolveTerminalWorkerSocketPath({ paseoHome: args.paseoHome, workerPid: process.pid, + recordId: args.recordId, }); if (process.platform !== "win32") { rmSync(socketPath, { force: true }); @@ -711,21 +716,11 @@ function runWorker(argv: string[]): void { chmodSync(socketPath, PRIVATE_FILE_MODE); } - const self = await inspector.inspect(process.pid); - if ( - self.status !== "alive" || - !self.snapshot.startedAt || - !self.snapshot.commandLine?.includes(args.recordId) - ) { - const reason = self.status === "error" ? errorMessage(self.error) : "no usable snapshot"; - throw new Error(`Terminal worker cannot read its own process identity: ${reason}`); - } record = { recordId: args.recordId, pid: process.pid, bootId: currentBootId(), createdAt: new Date().toISOString(), - identity: { startedAt: self.snapshot.startedAt, commandLine: self.snapshot.commandLine }, protocolVersion, paseoVersion: args.paseoVersion, socketPath, diff --git a/packages/server/src/terminal/terminal-worker-processes.ts b/packages/server/src/terminal/terminal-worker-processes.ts index b0c2e6d0cb..90de521bab 100644 --- a/packages/server/src/terminal/terminal-worker-processes.ts +++ b/packages/server/src/terminal/terminal-worker-processes.ts @@ -1,15 +1,13 @@ import { readFile } from "node:fs/promises"; -import { z } from "zod"; import { execCommand } from "../utils/spawn.js"; import { createSystemManagedProcessTable, type ManagedProcessCommandRunner, type ManagedProcessInspection, - windowsPowerShellPath, } from "../server/managed-processes/managed-processes.js"; -// Process inspection for terminal worker ownership. Birth times are compared as the strings -// `ps -o lstart` and Win32_Process.CreationDate print, so every reader must print them the same +// POSIX terminal program birth times are compared as the strings `ps -o lstart` prints, +// so every reader must print them the same // way: `ps` follows LC_TIME, and the worker, the daemon, the supervisor and the CLI can run // under different locales. Every `ps` here runs under LC_ALL=C. @@ -27,6 +25,7 @@ export interface ProcessTableEntry { export interface TerminalWorkerProcessInspector { /** True where the process table names each process's session (Linux). */ readonly reportsSessions: boolean; + readonly reportsBirthTimes: boolean; inspect(pid: number): Promise; /** One snapshot of every process, for walking descendants. Throws when it cannot be read. */ listProcesses(): Promise; @@ -37,14 +36,6 @@ export interface TerminalWorkerProcessInspector { readEnvironment(pid: number): Promise; } -const WindowsProcessListSchema = z.array( - z.object({ - ProcessId: z.number().int(), - ParentProcessId: z.number().int().nullable().optional(), - CreationDate: z.string().nullable().optional(), - }), -); - function createLocaleNeutralCommandRunner(): ManagedProcessCommandRunner { return { exec: (command, args) => @@ -65,11 +56,16 @@ export function createTerminalWorkerProcessInspector( const table = createSystemManagedProcessTable({ platform, commandRunner }); return { reportsSessions: platform === "linux", - inspect: (pid) => table.inspect(pid), - listProcesses: () => + reportsBirthTimes: platform !== "win32", + inspect: (pid) => platform === "win32" - ? listWindowsProcesses(commandRunner) - : listPosixProcesses(commandRunner, platform), + ? Promise.resolve({ + status: "error", + error: new Error("Windows terminal programs have no birth times"), + }) + : table.inspect(pid), + listProcesses: () => + platform === "win32" ? Promise.resolve([]) : listPosixProcesses(commandRunner, platform), readEnvironment: (pid) => readLinuxEnvironment(platform, pid), }; } @@ -121,33 +117,6 @@ async function listPosixProcesses( return entries; } -async function listWindowsProcesses( - commandRunner: ManagedProcessCommandRunner, -): Promise { - const command = - "@(Get-CimInstance Win32_Process | Select-Object ProcessId,ParentProcessId,CreationDate) | ConvertTo-Json -Compress"; - const { stdout } = await commandRunner.exec(windowsPowerShellPath(), [ - "-NoProfile", - "-NonInteractive", - "-Command", - command, - ]); - const parsed: unknown = JSON.parse(stdout.trim()); - const rows = WindowsProcessListSchema.parse(Array.isArray(parsed) ? parsed : [parsed]); - return rows.flatMap((row) => - row.CreationDate - ? [ - { - pid: row.ProcessId, - ppid: row.ParentProcessId ?? 0, - startedAt: row.CreationDate, - sessionId: null, - }, - ] - : [], - ); -} - /** * Every process below `rootPid` in the snapshot, plus (where the platform reports sessions) * every member of the session `rootPid` led and everything below those members. A PTY program diff --git a/packages/server/src/terminal/terminal-worker-reaper.test.ts b/packages/server/src/terminal/terminal-worker-reaper.test.ts index 66315b61d3..ef8d4fe487 100644 --- a/packages/server/src/terminal/terminal-worker-reaper.test.ts +++ b/packages/server/src/terminal/terminal-worker-reaper.test.ts @@ -22,6 +22,7 @@ import { terminalCleanupRecordPath, terminalWorkerDirectory, writeTerminalWorkerRecord, + writeTerminalCleanupRecord, type TerminalWorkerRecord, } from "./terminal-worker-record.js"; @@ -119,6 +120,7 @@ function withSessionMember(input: { }): TerminalWorkerProcessInspector { return { reportsSessions: true, + reportsBirthTimes: true, inspect: (pid) => inspector.inspect(pid), listProcesses: async () => { // A fresh snapshot each call, so it is rewritten in place. @@ -308,6 +310,7 @@ describe.skipIf(isPlatform("win32"))("terminal worker reaper", () => { const killable = spawnIdle(); const failing: TerminalWorkerProcessInspector = { reportsSessions: inspector.reportsSessions, + reportsBirthTimes: inspector.reportsBirthTimes, inspect: (pid) => pid === uninspectable.pid ? Promise.resolve({ status: "error", error: new Error("ps unavailable") }) @@ -504,6 +507,7 @@ describe.skipIf(isPlatform("win32"))("terminal worker reaper", () => { }); const tableUnavailable: TerminalWorkerProcessInspector = { reportsSessions: inspector.reportsSessions, + reportsBirthTimes: inspector.reportsBirthTimes, inspect: (pid) => inspector.inspect(pid), listProcesses: () => Promise.reject(new Error("ps timed out")), readEnvironment: (pid) => inspector.readEnvironment(pid), @@ -577,12 +581,19 @@ describe.skipIf(isPlatform("win32"))("terminal worker reaper", () => { const home = makeHome(); const worker = spawn( process.execPath, - ["-e", "setInterval(() => {}, 1e9)", "--", "--paseo-terminal-worker", "record-under-test"], + [ + "-e", + `require("node:net").createServer(() => {}).listen(${JSON.stringify(path.join(home, "live-record-under-test.sock"))});setInterval(() => {}, 1e9)`, + "--", + "--paseo-terminal-worker", + "record-under-test", + ], { detached: true, stdio: "ignore" }, ); children.push(worker); const recordPath = writeRecord(home, { pid: worker.pid!, + socketPath: path.join(home, "live-record-under-test.sock"), identity: { startedAt: await birthOf(worker.pid!), commandLine: "node -e --paseo-terminal-worker record-under-test", @@ -623,12 +634,19 @@ describe.skipIf(isPlatform("win32"))("terminal worker reaper", () => { const home = makeHome(); const worker = spawn( process.execPath, - ["-e", "setInterval(() => {}, 1e9)", "--", "--paseo-terminal-worker", "record-under-test"], + [ + "-e", + `require("node:net").createServer(() => {}).listen(${JSON.stringify(path.join(home, "live-record-under-test.sock"))});setInterval(() => {}, 1e9)`, + "--", + "--paseo-terminal-worker", + "record-under-test", + ], { detached: true, stdio: "ignore" }, ); children.push(worker); const recordPath = writeRecord(home, { pid: worker.pid!, + socketPath: path.join(home, "live-record-under-test.sock"), identity: { startedAt: await birthOf(worker.pid!), commandLine: "node -e --paseo-terminal-worker record-under-test", @@ -653,3 +671,128 @@ describe.skipIf(isPlatform("win32"))("terminal worker reaper", () => { }); }); }); + +// Real sockets identify workers on every platform, independently of program birth times. +describe("terminal worker reaper without process identity", () => { + const noIdentity = createTerminalWorkerProcessInspector(); + + // A live process standing in for a worker that cannot answer: it listens on the record's + // socket and never replies to the hello. + async function spawnSilentListener(socketPath: string): Promise { + const listener = spawnIdle(` + require("node:net").createServer(() => {}).listen(${JSON.stringify(socketPath)}, () => { + process.stdout.write("listening\\n"); + }); + `); + await new Promise((resolve) => listener.stdout!.once("data", resolve)); + return listener; + } + + it("retires the record of a gone worker and its gone programs without the identity query", async () => { + const home = makeHome(); + const recordPath = writeRecord(home, { + pid: await goneProcessPid(), + terminals: [{ id: "t1", pid: await goneProcessPid(), startedAt: null }], + }); + + const result = await reapTerminalWorkers({ paseoHome: home, inspector: noIdentity }); + + expect(result).toEqual({ records: 1, retired: 1, signaled: [] }); + expect(existsSync(recordPath)).toBe(false); + }); + + it("does not kill a newer legacy worker when a stale cleanup secret is rejected", async () => { + const home = makeHome(); + const socketPath = + process.platform === "win32" + ? `\\\\.\\pipe\\paseo-legacy-${Date.now()}` + : path.join(home, "legacy.sock"); + const listener = spawnIdle(` + require("node:net").createServer(socket => socket.on("data", () => socket.destroy())).listen(${JSON.stringify(socketPath)}, () => process.stdout.write("ready")); + `); + await new Promise((resolve) => listener.stdout!.once("data", resolve)); + const stalePath = writeRecord(home, { pid: listener.pid!, socketPath, secret: "stale-secret" }); + const stale = await readTerminalWorkerRecord(stalePath); + const cleanupPath = await writeTerminalCleanupRecord(home, { + ...stale, + closing: { + unresolved: [ + { pid: listener.pid!, role: "worker", reason: "still alive", detail: "old cleanup" }, + ], + }, + }); + writeRecord(home, { + pid: listener.pid!, + recordId: "newer-worker", + socketPath, + secret: "new-secret", + }); + await expect( + reapTerminalWorkers({ + paseoHome: home, + gracefulMs: 0, + confirmMs: 50, + recordPaths: [cleanupPath], + }), + ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError); + expect(isAlive(listener.pid!)).toBe(true); + }); + + it("force kills a worker whose socket accepts but never answers", async () => { + const home = makeHome(); + const socketPath = + process.platform === "win32" + ? `\\\\.\\pipe\\paseo-test-${Date.now()}-record-under-test` + : path.join(home, "w-record-under-test.sock"); + const listener = await spawnSilentListener(socketPath); + writeRecord(home, { + pid: listener.pid!, + socketPath, + }); + + const result = await reapTerminalWorkers({ + paseoHome: home, + inspector: noIdentity, + gracefulMs: 300, + }); + expect(result.signaled).toContain(listener.pid); + await waitUntilGone(listener.pid!); + expect(isAlive(listener.pid!)).toBe(false); + expect(await readUnconfirmedTerminalCleanup(home)).toEqual([]); + }); + + it("treats a live pid with nothing on the worker's socket as another process, signals it never, and retires the record", async () => { + const home = makeHome(); + const reusedWorkerPid = spawnIdle().pid!; + const recordPath = writeRecord(home, { + pid: reusedWorkerPid, + }); + + const result = await reapTerminalWorkers({ paseoHome: home, inspector: noIdentity }); + + expect(result).toEqual({ records: 1, retired: 1, signaled: [] }); + expect(isAlive(reusedWorkerPid)).toBe(true); + expect(existsSync(recordPath)).toBe(false); + }); + + it.skipIf(isPlatform("win32"))( + "never signals a live terminal program recorded without a birth time", + async () => { + const home = makeHome(); + const program = spawnIdle(); + writeRecord(home, { + pid: await goneProcessPid(), + terminals: [{ id: "t1", pid: program.pid!, startedAt: null }], + }); + + await expect( + reapTerminalWorkers({ paseoHome: home, inspector, confirmMs: 200 }), + ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError); + + expect(isAlive(program.pid!)).toBe(true); + expect(await readUnconfirmedTerminalCleanup(home)).toEqual([ + expect.objectContaining({ pid: program.pid!, role: "terminal program" }), + ]); + }, + ); +}); diff --git a/packages/server/src/terminal/terminal-worker-reaper.ts b/packages/server/src/terminal/terminal-worker-reaper.ts index 2647d4b25e..da229f9ddd 100644 --- a/packages/server/src/terminal/terminal-worker-reaper.ts +++ b/packages/server/src/terminal/terminal-worker-reaper.ts @@ -6,7 +6,8 @@ import { connectTerminalWorker } from "./terminal-worker-client.js"; import { pruneTerminalWorkerRuntimes } from "./terminal-worker-runtime.js"; import { classifyTerminalProgramProcess, - classifyTerminalWorkerProcess, + identifyTerminalWorker, + authorizeTerminalWorkerTermination, type ProcessOwnership, } from "./terminal-worker-ownership.js"; import { @@ -30,7 +31,8 @@ import { // uses for its own kill stop. Every kill intent (stop, quit, --force, a kill restart whose // shutdown hung, a dead or stopped worker, a program that ignores HUP) ends here: // -// 1. Classify the worker and every terminal program of each record on its own, by birth time. +// 1. Classify each record's worker by its pid and socket, and every terminal program on its +// own, by birth time (Windows records none; its programs end with their worker's console). // A record from an earlier boot names only processes the reboot ended: retire it unsignaled. // Descendants of owned programs (and, on Linux, members of their sessions) join the // candidates from one process-table snapshot taken before anything is signaled. Once a @@ -38,9 +40,10 @@ import { // program's session joins only when its environment carries that terminal's id. When the // table cannot be read, nothing in the record is signaled: ending a program would reparent // the children only the table can find, and a retry could never reach them again. -// 2. An owned worker gets an authenticated `shutdown`: it kills each terminal and waits for it. -// 3. After `gracefulMs`, SIGKILL whatever owned process is left, re-inspecting right before -// each signal. +// 2. A worker whose socket accepts is owned and gets an authenticated `shutdown`: it kills each +// terminal and waits for it. +// 3. After `gracefulMs`, SIGKILL whatever owned process is left, checking its socket or birth +// time again right before each signal. // 4. Succeed only when every candidate is gone or proven unrelated. Otherwise keep a record of // the unresolved entries, marked closing, in a cleanup record no worker rewrites, and throw // TerminalCleanupNotConfirmedError. An unreadable record is never reported confirmed. @@ -161,7 +164,7 @@ export interface ReapTerminalWorkersOptions { interface Candidate { pid: number; - startedAt: string; + startedAt: string | null; role: UnresolvedTerminalProcess["role"]; terminalId: string; state: ProcessOwnership; @@ -252,7 +255,7 @@ async function reapRecord( return { signaled: [], unresolved: [] }; } - const worker = classifyTerminalWorkerProcess(record, await inspector.inspect(record.pid)); + const worker = await identifyTerminalWorker(record); const discovery = await collectCandidates(record, inspector, log); const tracked: TrackedProcesses = { record, @@ -265,13 +268,11 @@ async function reapRecord( return await keepUnresolved(input, tracked, [], `not signaled: ${discovery.tableError}`); } - if (tracked.worker.status === "owned" && input.gracefulMs > 0) { - await requestWorkerShutdown(record, input.gracefulMs, log); + const shutdownMs = shutdownWindow(tracked.worker, input.gracefulMs); + if (shutdownMs > 0) { + await requestWorkerShutdown(record, shutdownMs, log); } - await waitUntilSettled( - tracked, - Date.now() + (tracked.worker.status === "owned" ? input.gracefulMs : 0), - ); + await waitUntilSettled(tracked, Date.now() + shutdownMs); let signaled: number[] = []; if (!isSettled(tracked)) { @@ -282,6 +283,13 @@ async function reapRecord( return await keepUnresolved(input, tracked, signaled, null); } +// Only an accepting socket authorizes shutdown and eventual forced termination. +function shutdownWindow(worker: ProcessOwnership, gracefulMs: number): number { + if (worker.status === "owned") return gracefulMs; + + return 0; +} + // Retires the record when every process it names is settled; otherwise writes the cleanup // record, marked closing, with only the unresolved entries. A live worker keeps rewriting its // own record, so the closing state never goes there. `notSignaled` explains open processes that @@ -337,10 +345,7 @@ function isSettled(tracked: TrackedProcesses): boolean { async function refreshTracked(tracked: TrackedProcesses): Promise { if (isOpen(tracked.worker)) { - tracked.worker = classifyTerminalWorkerProcess( - tracked.record, - await tracked.inspector.inspect(tracked.record.pid), - ); + tracked.worker = await identifyTerminalWorker(tracked.record); } for (const candidate of tracked.candidates) { if (isOpen(candidate.state)) { @@ -369,7 +374,7 @@ async function forceKillOwned( const { record, inspector } = tracked; const signaled: number[] = []; if (tracked.worker.status === "owned") { - tracked.worker = classifyTerminalWorkerProcess(record, await inspector.inspect(record.pid)); + tracked.worker = await authorizeTerminalWorkerTermination(record); if (tracked.worker.status === "owned") { await forceKillWorker(record.pid); signaled.push(record.pid); @@ -447,6 +452,7 @@ async function collectCandidates( log: NonNullable, ): Promise { const candidates = new Map(); + if (inspector.reportsBirthTimes === false) return { candidates: [], tableError: null }; for (const entry of record.terminals) { candidates.set(entry.pid, { pid: entry.pid, @@ -461,8 +467,12 @@ async function collectCandidates( // already gone: the kernel keeps a session id reserved while a member lives, so a job the // shell left behind is still found. A pid reused by another process (unrelated) leads someone // else's session and is never walked. + // A program recorded without a birth time is never walked: the worker could not read the + // process table when it started it, so the walk would only keep the cleanup unconfirmed. const roots = Array.from(candidates.values()).filter( - (c) => c.state.status === "owned" || (c.state.status === "gone" && inspector.reportsSessions), + (c) => + c.state.status === "owned" || + (c.state.status === "gone" && c.startedAt !== null && inspector.reportsSessions), ); if (roots.length === 0) return { candidates: Array.from(candidates.values()), tableError: null }; @@ -564,7 +574,7 @@ async function requestWorkerShutdown( timeoutMs: Math.min(gracefulMs, 2_000), }); } catch (error) { - // A worker that cannot be reached is escalated by birth time below; a failed connection + // A worker that cannot be reached is probed again below; a failed authentication // alone never authorizes a signal. log("Terminal worker did not accept a shutdown request", { pid: record.pid, diff --git a/packages/server/src/terminal/terminal-worker-record.ts b/packages/server/src/terminal/terminal-worker-record.ts index 3176d2ee8e..63ac0ddd91 100644 --- a/packages/server/src/terminal/terminal-worker-record.ts +++ b/packages/server/src/terminal/terminal-worker-record.ts @@ -4,14 +4,20 @@ import path from "node:path"; import { z } from "zod"; import { writePrivateFileAtomicSync, writePrivateFileAtomic } from "../server/private-files.js"; -// Owner-only metadata a detached terminal worker leaves in $PASEO_HOME so the next daemon can -// find, authenticate and adopt it, and so cleanup can identify every process it is allowed to -// signal. Process identity is the birth time `ps`/Win32_Process report, never the pid alone. +// Owner-only metadata used to find and authenticate workers in the locked home. const TerminalWorkerTerminalEntrySchema = z.object({ id: z.string().min(1), pid: z.number().int().positive(), + // Null when the program's birth time could not be read: it is never signaled. + startedAt: z.string().min(1).nullable(), +}); + +// The birth time and command line workers up to 0.15.1 recorded about themselves. Still read +// from their records, never written, and never consulted: the socket identifies a worker. +const LegacyTerminalWorkerIdentitySchema = z.object({ startedAt: z.string().min(1), + commandLine: z.string().min(1), }); // One process a cleanup could not confirm gone, kept so a later caller (a retry, or a remote @@ -28,10 +34,7 @@ const TerminalWorkerRecordSchema = z.object({ pid: z.number().int().positive(), bootId: z.string().nullable(), createdAt: z.string().min(1), - identity: z.object({ - startedAt: z.string().min(1), - commandLine: z.string().min(1), - }), + identity: LegacyTerminalWorkerIdentitySchema.optional(), // Read loosely: a version this daemon cannot interpret marks a worker from a newer Paseo // (classifyTerminalWorkerProtocol), not an unreadable record. protocolVersion: z.union([z.number(), z.string()]), diff --git a/packages/server/src/terminal/terminal-worker-socket-path.test.ts b/packages/server/src/terminal/terminal-worker-socket-path.test.ts index 409cf60c81..6017177fcf 100644 --- a/packages/server/src/terminal/terminal-worker-socket-path.test.ts +++ b/packages/server/src/terminal/terminal-worker-socket-path.test.ts @@ -37,6 +37,7 @@ describe.skipIf(isPlatform("win32"))("terminal worker socket path", () => { const home = makeDir("tw-h-", "/tmp"); const socketPath = resolveTerminalWorkerSocketPath({ paseoHome: home, + recordId: "test", workerPid: 4242, platform: "darwin", }); @@ -51,6 +52,7 @@ describe.skipIf(isPlatform("win32"))("terminal worker socket path", () => { const uid = process.getuid!(); const socketPath = resolveTerminalWorkerSocketPath({ paseoHome: home, + recordId: "test", workerPid: 99999, platform: "darwin", tmpDir, @@ -71,6 +73,7 @@ describe.skipIf(isPlatform("win32"))("terminal worker socket path", () => { // also be refused, and /tmp/paseo- takes over. const socketPath = resolveTerminalWorkerSocketPath({ paseoHome: home, + recordId: "test", workerPid: 7, platform: "darwin", tmpDir, @@ -84,6 +87,7 @@ describe.skipIf(isPlatform("win32"))("terminal worker socket path", () => { path.dirname( resolveTerminalWorkerSocketPath({ paseoHome: home, + recordId: "test", workerPid: 7, platform: "darwin", tmpDir: wideTmp, @@ -101,6 +105,7 @@ describe.skipIf(isPlatform("win32"))("terminal worker socket path", () => { try { resolveTerminalWorkerSocketPath({ paseoHome: home, + recordId: "test", workerPid: 7, platform: "darwin", tmpDir, @@ -123,7 +128,11 @@ describe.skipIf(isPlatform("win32"))("terminal worker socket path", () => { it("accepts a real listen on the chosen path", async () => { const home = makeDir("tw-l-", "/tmp"); - const socketPath = resolveTerminalWorkerSocketPath({ paseoHome: home, workerPid: 5 }); + const socketPath = resolveTerminalWorkerSocketPath({ + paseoHome: home, + recordId: "test", + workerPid: 5, + }); const server = createServer(); await new Promise((resolve, reject) => { server.once("error", reject); @@ -132,3 +141,10 @@ describe.skipIf(isPlatform("win32"))("terminal worker socket path", () => { await new Promise((resolve) => server.close(() => resolve())); }); }); + +it("uses a separate socket for each record even when its pid is reused", () => { + const paseoHome = makeDir("tw-id-"); + const first = resolveTerminalWorkerSocketPath({ paseoHome, workerPid: 42, recordId: "old" }); + const second = resolveTerminalWorkerSocketPath({ paseoHome, workerPid: 42, recordId: "new" }); + expect(first).not.toBe(second); +}); diff --git a/packages/server/src/terminal/terminal-worker-socket-path.ts b/packages/server/src/terminal/terminal-worker-socket-path.ts index 99fce0e64c..d9a153c598 100644 --- a/packages/server/src/terminal/terminal-worker-socket-path.ts +++ b/packages/server/src/terminal/terminal-worker-socket-path.ts @@ -18,6 +18,7 @@ const OWNER_ONLY_MODE = 0o700; export interface TerminalWorkerSocketPathInput { paseoHome: string; workerPid: number; + recordId: string; platform?: NodeJS.Platform; tmpDir?: string; uid?: number; @@ -60,7 +61,7 @@ export function terminalWorkerSocketId(paseoHome: string, workerPid: number): st */ export function resolveTerminalWorkerSocketPath(input: TerminalWorkerSocketPathInput): string { const platform = input.platform ?? process.platform; - const id = terminalWorkerSocketId(input.paseoHome, input.workerPid); + const id = terminalWorkerSocketId(input.paseoHome, input.workerPid) + "-" + input.recordId; if (platform === "win32") { return `\\\\.\\pipe\\paseo-tw-${id}`; } diff --git a/packages/server/src/terminal/worker-terminal-manager.test.ts b/packages/server/src/terminal/worker-terminal-manager.test.ts index 759f73e639..26cbb031d0 100644 --- a/packages/server/src/terminal/worker-terminal-manager.test.ts +++ b/packages/server/src/terminal/worker-terminal-manager.test.ts @@ -62,15 +62,22 @@ import { readTerminalWorkerRecord, terminalWorkerDirectory, terminalWorkerRecordPath, + writeTerminalWorkerRecord, + writeTerminalCleanupRecord, type TerminalWorkerRecord, } from "./terminal-worker-record.js"; -import { TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV } from "./terminal-worker-protocol.js"; +import { + TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV, + TERMINAL_WORKER_PROTOCOL_VERSION, +} from "./terminal-worker-protocol.js"; import { NoTerminalWorkerAvailableError, TerminalWorkerIncompatibleError, } from "./terminal-worker-pool.js"; import { connectTerminalWorker } from "./terminal-worker-client.js"; import { launchTerminalWorker, TerminalWorkerLaunchError } from "./terminal-worker-launch.js"; +import { currentBootId } from "../server/pid-lock.js"; +import { spawn } from "node:child_process"; type TerminalRow = TerminalState["grid"][number]; @@ -440,26 +447,24 @@ it("creates a terminal through the worker and streams output", async () => { }), ); const messages: string[] = []; - let readinessEvents = 0; + let snapshotReady = false; const unsubscribe = session.subscribe((message) => { if (message.type === "output") { messages.push(message.data); } if (message.type === "snapshotReady") { - readinessEvents += 1; + snapshotReady = true; } }); - await new Promise((resolve) => setTimeout(resolve, 100)); - expect(readinessEvents).toBe(1); + // Subscription and initial PTY output each announce readiness; their order depends on startup. + await waitForCondition(() => snapshotReady, 10000); session.send({ type: "input", data: "hello\r" }); await waitForCondition(() => messages.join("").includes("worker-output:hello"), 10000); - await new Promise((resolve) => setTimeout(resolve, 100)); unsubscribe(); expect(messages.join("")).toContain("worker-output:hello"); - expect(readinessEvents).toBe(1); }); it("delivers rapid small writes complete and in order through worker coalescing", async () => { @@ -812,7 +817,7 @@ async function recordedTerminalPids(paseoHome: string): Promise { const paseoHome = createTemporaryHome(); const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-unfinished-kill-")); @@ -844,17 +849,22 @@ it.each([false, true])( const remainingPid = pids.get(remaining.id)!; const [record] = await workerRecords(paseoHome); - // A kill cleanup whose process table read fails signals nothing and leaves the worker live. + // Seed a retained cleanup through the same record interface the reaper writes. Windows + // does not read program birth times, so a process-table failure cannot create this state. const inspector = createTerminalWorkerProcessInspector(); - await expect( - reapTerminalWorkers({ - paseoHome, - inspector: { - ...inspector, - listProcesses: () => Promise.reject(new Error("ps timed out")), - }, - }), - ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError); + await writeTerminalCleanupRecord(paseoHome, { + ...record!, + closing: { + unresolved: [ + { + pid: record!.pid, + role: "worker", + reason: "still alive", + detail: "shutdown was not confirmed", + }, + ], + }, + }); expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0); // Keep an ordinary reader open until the worker delivers the terminal exit. @@ -1643,17 +1653,16 @@ it.each([1, 4])( expect(await readUnconfirmedTerminalCleanup(paseoHome)).toEqual([]); // Startup launches a replacement worker after cleanup. Either old PID may already // belong to another process, so verify that the recorded birth identities are gone. - for (const original of [{ pid: record!.pid, ...record!.identity }, program]) { - await expect - .poll( - async () => - !isProcessAlive(original.pid) || - classifyTerminalProgramProcess(original, await inspector.inspect(original.pid)) - .status === "unrelated", - { timeout: 10000 }, - ) - .toBe(true); - } + await waitForCondition(() => !isProcessAlive(record!.pid), 10000); + await expect + .poll( + async () => + !isProcessAlive(program.pid) || + classifyTerminalProgramProcess(program, await inspector.inspect(program.pid)).status === + "unrelated", + { timeout: 10000 }, + ) + .toBe(true); expect(manager.getTerminal(old.terminalId)).toBeUndefined(); const workers = await workerRecords(paseoHome); expect(workers).toHaveLength(1); @@ -2683,33 +2692,216 @@ it.each([false, true])( 45000, ); -// Both Windows reports in #8766: the worker's process identity query (PowerShell there, `ps` here) -// fails, the worker exits before writing its record, and the daemon used to exit with it. -it.skipIf(isPlatform("win32"))( - "starts without a worker when the worker cannot read its own identity, and launches one once it can", +// #8810: a worker that cannot start must not take the daemon down. An override that is not a +// protocol version makes the worker fail before it writes its record, on every platform. +it("starts without a worker when the worker cannot start, and launches one once it can", async () => { + const paseoHome = createTemporaryHome(); + const previous = process.env[TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV]; + process.env[TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV] = "not-a-version"; + try { + manager = await startRealWorkerManager({ paseoHome }); + expect(await workerRecords(paseoHome)).toEqual([]); + await expect( + manager.createTerminal({ workspaceId: "ws-no-worker", cwd: paseoHome }), + ).rejects.toThrow(TerminalWorkerLaunchError); + } finally { + if (previous === undefined) delete process.env[TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV]; + else process.env[TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV] = previous; + } + + trackTerminal(await manager.createTerminal({ workspaceId: "ws-no-worker", cwd: paseoHome })); + expect(await workerRecords(paseoHome)).toHaveLength(1); +}, 30000); + +function echoTerminalCommand(prefix: string) { + return nodeTerminalCommand(` + process.stdin.on("data", (chunk) => { + process.stdout.write(${JSON.stringify(prefix)} + chunk.toString()); + }); + setInterval(() => {}, 1000); + `); +} + +async function expectEcho(session: TerminalSession, prefix: string, input: string): Promise { + const messages: string[] = []; + const unsubscribe = session.subscribe((message) => { + if (message.type === "output") messages.push(message.data); + }); + session.send({ type: "input", data: `${input}\r` }); + try { + await waitForCondition(() => messages.join("").includes(`${prefix}${input}`), 10000); + } finally { + unsubscribe(); + } +} + +it("runs terminals without recording a worker process identity", async () => { + const paseoHome = createTemporaryHome(); + manager = await startRealWorkerManager({ paseoHome }); + const session = trackTerminal( + await manager.createTerminal({ + workspaceId: "ws-no-identity", + cwd: paseoHome, + ...echoTerminalCommand("no-identity:"), + }), + ); + await expectEcho(session, "no-identity:", "hello"); + + const [record, ...others] = await workerRecords(paseoHome); + expect(others).toEqual([]); + expect(record!.identity).toBeUndefined(); + expect(record!.terminals).toEqual([ + { + id: session.id, + pid: expect.any(Number), + startedAt: process.platform === "win32" ? null : expect.any(String), + }, + ]); +}, 30000); + +it("adopts terminals over the authenticated socket after a daemon restart", async () => { + const paseoHome = createTemporaryHome(); + const first = await startRealWorkerManager({ paseoHome }); + const created = await first.createTerminal({ + workspaceId: "ws-no-identity", + cwd: paseoHome, + ...echoTerminalCommand("adopted:"), + }); + await first.detach(); + + manager = await startRealWorkerManager({ paseoHome }); + + const adopted = manager.getTerminal(created.id); + expect(adopted).toBeDefined(); + await expectEcho(adopted!, "adopted:", "after-restart"); + expect(await workerRecords(paseoHome)).toHaveLength(1); +}, 30000); + +it("closes terminals and their worker through authenticated shutdown", async () => { + const paseoHome = createTemporaryHome(); + const stopping = await startRealWorkerManager({ paseoHome }); + manager = stopping; + await stopping.createTerminal({ + workspaceId: "ws-no-identity", + cwd: paseoHome, + ...nodeTerminalCommand("setInterval(() => {}, 1000);"), + }); + const [record] = await workerRecords(paseoHome); + const programPid = record!.terminals[0]!.pid; + + await stopping.closeTerminals(); + + // The worker closed its terminals before it answered; it stops listening as it exits. + expect(isProcessAlive(programPid)).toBe(false); + expect(await listTerminalWorkerRecords(paseoHome)).toEqual([]); + await waitForCondition(() => !isProcessAlive(record!.pid), 5000); +}, 30000); + +// Startup cleanup and retries can request immediate forced termination. +it("force closes a worker from cleanup without a grace period", async () => { + const paseoHome = createTemporaryHome(); + const previous = await startRealWorkerManager({ paseoHome }); + await previous.createTerminal({ + workspaceId: "ws-no-identity", + cwd: paseoHome, + ...nodeTerminalCommand("setInterval(() => {}, 1000);"), + }); + const [record] = await workerRecords(paseoHome); + const programPid = record!.terminals[0]!.pid; + await previous.detach(); + + const result = await reapTerminalWorkers({ paseoHome, gracefulMs: 0 }); + + expect(result.signaled).toContain(record!.pid); + expect(isProcessAlive(programPid)).toBe(false); + expect(await listTerminalWorkerRecords(paseoHome)).toEqual([]); + await waitForCondition(() => !isProcessAlive(record!.pid), 5000); +}, 30000); + +it("launches past stale records without signaling an unrelated live pid", async () => { + const paseoHome = createTemporaryHome(); + // One worker died; another died and its pid now runs an unrelated process, so nothing + // listens on its socket. Neither may count against the two-worker limit or be signaled. + const exited = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await new Promise((resolve) => exited.once("exit", resolve)); + const reused = spawn(process.execPath, ["-e", "setInterval(() => {}, 1e9)"], { + stdio: "ignore", + }); + try { + for (const pid of [exited.pid!, reused.pid!]) { + writeTerminalWorkerRecord(paseoHome, { + recordId: `dead-worker-${pid}`, + pid, + bootId: currentBootId(), + createdAt: new Date().toISOString(), + protocolVersion: TERMINAL_WORKER_PROTOCOL_VERSION, + paseoVersion: "0.0.0-test", + socketPath: join(paseoHome, `gone-${pid}.sock`), + secret: "secret-of-a-gone-worker", + terminals: [], + }); + } + manager = await startRealWorkerManager({ paseoHome }); + trackTerminal(await manager.createTerminal({ workspaceId: "ws-no-identity", cwd: paseoHome })); + + const records = await workerRecords(paseoHome); + expect(records.map((record) => record.pid)).not.toContain(exited.pid); + expect(records.map((record) => record.pid)).not.toContain(reused.pid); + expect(records).toHaveLength(1); + expect(isProcessAlive(reused.pid!)).toBe(true); + } finally { + reused.kill("SIGKILL"); + } +}, 30000); + +it("does not reach a newer worker through a stale record with its reused pid", async () => { + const paseoHome = createTemporaryHome(); + manager = await startRealWorkerManager({ paseoHome }); + const session = trackTerminal( + await manager.createTerminal({ + workspaceId: "stale-worker-record", + cwd: paseoHome, + ...echoTerminalCommand("alive:"), + }), + ); + const [record] = await workerRecords(paseoHome); + const staleHome = createTemporaryHome(); + writeTerminalWorkerRecord(staleHome, { + ...record!, + recordId: "stale-record", + socketPath: + process.platform === "win32" + ? `\\\\.\\pipe\\paseo-stale-${Date.now()}` + : join(staleHome, "gone.sock"), + terminals: [], + }); + expect(await reapTerminalWorkers({ paseoHome: staleHome, gracefulMs: 0 })).toMatchObject({ + retired: 1, + signaled: [], + }); + await expectEcho(session, "alive:", "untouched"); +}, 30000); + +it.skipIf(!isPlatform("win32"))( + "observes terminal program survival after a worker hard crash", async () => { const paseoHome = createTemporaryHome(); - const binDir = mkdtempSync(join(tmpdir(), "worker-terminal-manager-no-ps-")); - temporaryDirs.push(binDir); - writeFileSync(join(binDir, "ps"), "#!/bin/sh\nexit 1\n", { mode: 0o755 }); - const originalPath = process.env.PATH; - process.env.PATH = `${binDir}${delimiter}${originalPath ?? ""}`; + const previous = await startRealWorkerManager({ paseoHome }); + await previous.createTerminal({ + workspaceId: "worker-crash", + cwd: paseoHome, + ...nodeTerminalCommand("setInterval(() => {}, 1000)"), + }); + const [record] = await workerRecords(paseoHome); + await previous.detach(); + process.kill(record!.pid, "SIGKILL"); + await waitForCondition(() => !isProcessAlive(record!.pid), 5000); + const program = record!.terminals[0]!.pid; try { - manager = await startRealWorkerManager({ paseoHome }); - expect(await workerRecords(paseoHome)).toEqual([]); - await expect( - manager.createTerminal({ workspaceId: "ws-no-identity", cwd: paseoHome }), - ).rejects.toThrow(TerminalWorkerLaunchError); - expect(await workerRecords(paseoHome)).toEqual([]); - } finally { - process.env.PATH = originalPath; - } - - const terminal = trackTerminal( - await manager.createTerminal({ workspaceId: "ws-no-identity", cwd: paseoHome }), - ); - expect(terminal.id).toBeTruthy(); - expect((await workerRecords(paseoHome)).map((record) => record.pid)).toHaveLength(1); + await waitForCondition(() => !isProcessAlive(program), 5000); + } catch {} + console.log(`Windows orphaned-program check: ${isProcessAlive(program) ? "survived" : "died"}`); + if (isProcessAlive(program)) process.kill(program, "SIGKILL"); }, 30000, ); diff --git a/packages/server/src/terminal/worker-terminal-manager.ts b/packages/server/src/terminal/worker-terminal-manager.ts index ea91d54a66..2592b0c89c 100644 --- a/packages/server/src/terminal/worker-terminal-manager.ts +++ b/packages/server/src/terminal/worker-terminal-manager.ts @@ -21,11 +21,8 @@ import { resolvePaseoCliExecutablePath } from "./paseo-cli-path.js"; import type { CaptureTerminalLinesResult } from "./terminal-capture.js"; import { connectTerminalWorker, type TerminalWorkerConnection } from "./terminal-worker-client.js"; import { launchTerminalWorker } from "./terminal-worker-launch.js"; -import { classifyTerminalWorkerProcess } from "./terminal-worker-ownership.js"; -import { - createTerminalWorkerProcessInspector, - type TerminalWorkerProcessInspector, -} from "./terminal-worker-processes.js"; +import { identifyTerminalWorker } from "./terminal-worker-ownership.js"; +import { createTerminalWorkerProcessInspector } from "./terminal-worker-processes.js"; import { reapTerminalWorkers } from "./terminal-worker-reaper.js"; import { listTerminalWorkerRecords, @@ -1038,18 +1035,21 @@ export async function startWorkerTerminalManager( toReap.push(file.path); continue; } - const worker = classifyTerminalWorkerProcess(record, await inspector.inspect(record.pid)); - if (worker.status === "owned") { + const worker = await identifyTerminalWorker(record); + if (worker.status === "owned" || worker.status === "unknown") { + if (worker.status === "unknown") { + // Its secret still proves it: only the worker that wrote the record can answer the + // hello below. An inconclusive probe does not authorize a signal. + logger.warn( + { recordPath: file.path, pid: record.pid, reason: worker.reason }, + "Terminal worker socket probe inconclusive; attempting authentication", + ); + } live.push({ path: file.path, record, fit: classifyTerminalWorkerProtocol(record.protocolVersion), }); - } else if (worker.status === "unknown") { - logger.warn( - { recordPath: file.path, pid: record.pid, reason: worker.reason }, - "Cannot identify terminal worker; leaving its record", - ); } else { // Its programs lost their PTY with the worker and cannot be adopted. toReap.push(file.path); @@ -1142,10 +1142,7 @@ export async function startWorkerTerminalManager( const reconnect = () => { recovery ??= (async () => { while (!recoveryState.stopped) { - const ownership = classifyTerminalWorkerProcess( - record, - await inspector.inspect(record.pid), - ); + const ownership = await identifyTerminalWorker(record); if (ownership.status === "gone" || ownership.status === "unrelated") { await reapTerminalWorkers({ paseoHome: options.paseoHome, @@ -1243,7 +1240,7 @@ export async function startWorkerTerminalManager( // Counted from the records rather than from what startup drove: a worker a cleanup could not // confirm gone keeps running, and a draining one exits on its own. const startCurrent = async (): Promise => { - const running = await countRunningTerminalWorkers(options.paseoHome, inspector); + const running = await countRunningTerminalWorkers(options.paseoHome); if (running >= MAX_TERMINAL_WORKERS) { throw new NoTerminalWorkerAvailableError( `${running} terminal workers are already running in ${options.paseoHome}; new terminals can open once one of them exits`, @@ -1273,8 +1270,8 @@ export async function startWorkerTerminalManager( logger.warn(error.message); } else { // The daemon starts without terminals rather than not at all. A worker that cannot start - // here (its process identity query failed, say) is launched again by the first terminal - // creation, which reports the same error if it still cannot. + // here is launched again by the first terminal creation, which reports the same error + // if it still cannot. logger.error( { err: error }, "Terminal worker did not start; terminals are unavailable until one starts", @@ -1298,11 +1295,9 @@ export async function startWorkerTerminalManager( // Every worker that may still run in this home, one per record: a worker and its cleanup record // are listed as one. Unreadable records and workers that cannot be identified count; a worker -// confirmed gone, or a pid reused by another process, does not. -async function countRunningTerminalWorkers( - paseoHome: string, - inspector: TerminalWorkerProcessInspector, -): Promise { +// confirmed gone, or a pid now running something else (another birth time, or nothing listening +// on the worker's socket), does not. +async function countRunningTerminalWorkers(paseoHome: string): Promise { let running = 0; for (const file of await listTerminalWorkerRecords(paseoHome)) { if ("error" in file) { @@ -1311,7 +1306,7 @@ async function countRunningTerminalWorkers( } const { record } = file; if (!writtenDuringThisBoot({ startedAt: record.createdAt, bootId: record.bootId })) continue; - const worker = classifyTerminalWorkerProcess(record, await inspector.inspect(record.pid)); + const worker = await identifyTerminalWorker(record); if (worker.status === "owned" || worker.status === "unknown") running += 1; } return running;