From b0e6dbceef58fbabb422ff8e74741a679e655220 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 20 Jun 2026 22:05:12 +0800 Subject: [PATCH] Clean up owned helper processes on daemon startup (#1632) Record provider-owned helper processes (currently the OpenCode `serve` helper) in a daemon ledger and reconcile it in the background on startup: terminate validated leftovers, drop dead or PID-reused records without killing anything, and keep a record whose process can't be inspected for the next reconcile. Termination stops as soon as the process exits instead of always escalating to SIGKILL. --- docs/data-model.md | 3 + docs/providers.md | 6 + .../src/server/agent/provider-registry.ts | 19 +- .../server/agent/provider-snapshot-manager.ts | 5 + .../server/agent/providers/opencode-agent.ts | 6 +- .../providers/opencode-server-manager.test.ts | 308 ++++++++---- .../providers/opencode/server-manager.ts | 159 ++++-- .../bootstrap-managed-processes.test.ts | 90 ++++ packages/server/src/server/bootstrap.ts | 41 ++ .../managed-processes.test.ts | 391 +++++++++++++++ .../managed-processes/managed-processes.ts | 455 ++++++++++++++++++ 11 files changed, 1359 insertions(+), 124 deletions(-) create mode 100644 packages/server/src/server/bootstrap-managed-processes.test.ts create mode 100644 packages/server/src/server/managed-processes/managed-processes.test.ts create mode 100644 packages/server/src/server/managed-processes/managed-processes.ts diff --git a/docs/data-model.md b/docs/data-model.md index 38e0c8f6c..fe90e475d 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -27,6 +27,9 @@ $PASEO_HOME/ ├── projects/ │ ├── projects.json # Project registry │ └── workspaces.json # Workspace registry +├── runtime/ +│ └── managed-processes/ +│ └── {recordId}.json # Helper processes owned by Paseo; reconciled on daemon bootstrap └── push-tokens.json # Expo push notification tokens ``` diff --git a/docs/providers.md b/docs/providers.md index 3f41aba6e..db0f026de 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -38,6 +38,12 @@ Draft metadata lookups should avoid creating provider sessions when the upstream Provider session import has its own contract. The picker calls `listImportableSessions` and receives rows only: provider handle, cwd, title, prompt previews, and last activity. Import calls `importSession({ providerHandleId, cwd })` for the selected row and must not call listing again. The provider returns the resumed session, storage config, persistence handle, and hydrated timeline for that one native session; `AgentManager.importProviderSession` seeds the daemon timeline and publishes the Paseo agent only after it is ready. +## Provider Helper Processes + +Provider-owned helper processes that can outlive an individual agent session must be recorded in the daemon's managed-process registry. Store provider/kind metadata, the PID, launch command/args, and process identity captured from the platform process table. Remove the record on normal exit or shutdown. + +Daemon bootstrap reconciles that ledger in the background, without blocking startup: dead PIDs are deleted, PID identity mismatches are deleted without killing anything, only positively matched Paseo-owned leftovers are terminated, and a record whose process cannot be inspected is left in place for the next reconcile rather than deleted. Do not add broad process-name sweepers for provider cleanup; cleanup starts from records Paseo previously wrote. + --- ## Provider Snapshot Refresh Contract diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index 85f3ae5bc..9ecf23329 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -21,6 +21,7 @@ import { } from "./create-agent-mode.js"; import { normalizeAgentModelDefinition } from "./agent-sdk-types.js"; import type { WorkspaceGitService } from "../workspace-git-service.js"; +import type { ManagedProcessRegistry } from "../managed-processes/managed-processes.js"; import type { AgentProviderRuntimeSettingsMap, ProviderOverride, @@ -71,12 +72,13 @@ export interface BuildProviderRegistryOptions { runtimeSettings?: AgentProviderRuntimeSettingsMap; providerOverrides?: Record; workspaceGitService?: Pick; + managedProcesses?: ManagedProcessRegistry; isDev?: boolean; } interface ProviderClientFactoryOptions extends Pick< BuildProviderRegistryOptions, - "workspaceGitService" + "workspaceGitService" | "managedProcesses" > { providerParams?: unknown; customProvider?: { @@ -126,7 +128,10 @@ const PROVIDER_CLIENT_FACTORIES: Record = { command: getCursorACPCommand(runtimeSettings), env: runtimeSettings?.env, }), - opencode: (logger, runtimeSettings) => new OpenCodeAgentClient(logger, runtimeSettings), + opencode: (logger, runtimeSettings, options) => + new OpenCodeAgentClient(logger, runtimeSettings, { + managedProcesses: options?.managedProcesses, + }), pi: (logger, runtimeSettings, options) => new PiRpcAgentClient({ logger, @@ -528,7 +533,7 @@ function createResolvedProviderClient( function buildResolvedBuiltinProviders( providerOverrides: Record, runtimeSettings: AgentProviderRuntimeSettingsMap | undefined, - options: Pick, + options: Pick, isDev: boolean, ): Map { const resolvedProviders = new Map(); @@ -557,6 +562,7 @@ function buildResolvedBuiltinProviders( createBaseClient: (logger) => factory(logger, mergedRuntimeSettings, { workspaceGitService: options.workspaceGitService, + managedProcesses: options.managedProcesses, providerParams: override?.params, }), }); @@ -568,6 +574,7 @@ function buildResolvedBuiltinProviders( function addDerivedProviders( resolvedProviders: Map, providerOverrides: Record, + options: Pick, ): void { for (const [providerId, override] of Object.entries(providerOverrides)) { if (resolvedProviders.has(providerId) || BUILTIN_PROVIDER_IDS.includes(providerId)) { @@ -653,6 +660,7 @@ function addDerivedProviders( providerParams, createBaseClient: (logger) => baseFactory(logger, mergedRuntimeSettings, { + managedProcesses: options.managedProcesses, providerParams, customProvider: { id: providerId, @@ -675,10 +683,13 @@ export function buildProviderRegistry( runtimeSettings, { workspaceGitService: options?.workspaceGitService, + managedProcesses: options?.managedProcesses, }, options?.isDev === true, ); - addDerivedProviders(resolvedProviders, providerOverrides); + addDerivedProviders(resolvedProviders, providerOverrides, { + managedProcesses: options?.managedProcesses, + }); return Object.fromEntries( [...resolvedProviders.entries()].map(([provider, resolved]) => [ diff --git a/packages/server/src/server/agent/provider-snapshot-manager.ts b/packages/server/src/server/agent/provider-snapshot-manager.ts index e0d1c1add..054fcd806 100644 --- a/packages/server/src/server/agent/provider-snapshot-manager.ts +++ b/packages/server/src/server/agent/provider-snapshot-manager.ts @@ -16,6 +16,7 @@ import type { } from "./agent-sdk-types.js"; import type { ManagedAgent } from "./agent-manager.js"; import type { WorkspaceGitService } from "../workspace-git-service.js"; +import type { ManagedProcessRegistry } from "../managed-processes/managed-processes.js"; import type { AgentProviderRuntimeSettingsMap, ProviderOverride, @@ -56,6 +57,7 @@ export interface ProviderSnapshotManagerOptions { runtimeSettings?: AgentProviderRuntimeSettingsMap; providerOverrides?: Record; workspaceGitService?: Pick; + managedProcesses?: ManagedProcessRegistry; isDev?: boolean; extraClients?: Partial>; refreshTimeoutMs?: number; @@ -127,6 +129,7 @@ export class ProviderSnapshotManager { private readonly refreshTimeoutMs: number; private readonly logger: Logger; private readonly workspaceGitService?: Pick; + private readonly managedProcesses?: ManagedProcessRegistry; private readonly isDev: boolean; private readonly extraClients: Partial>; private runtimeSettings: AgentProviderRuntimeSettingsMap | undefined; @@ -138,6 +141,7 @@ export class ProviderSnapshotManager { constructor(options: ProviderSnapshotManagerOptions) { this.logger = options.logger; this.workspaceGitService = options.workspaceGitService; + this.managedProcesses = options.managedProcesses; this.isDev = options.isDev === true; this.extraClients = options.extraClients ?? {}; this.runtimeSettings = options.runtimeSettings; @@ -370,6 +374,7 @@ export class ProviderSnapshotManager { runtimeSettings: this.runtimeSettings, providerOverrides: this.providerOverrides, workspaceGitService: this.workspaceGitService, + managedProcesses: this.managedProcesses, isDev: this.isDev, }); diff --git a/packages/server/src/server/agent/providers/opencode-agent.ts b/packages/server/src/server/agent/providers/opencode-agent.ts index 3b577bcb2..24d35efab 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.ts @@ -84,6 +84,7 @@ import { } from "./opencode/runtime.js"; import { normalizeProviderReplayTimestamp } from "../provider-history-timestamps.js"; import { revertOpenCodeConversationAndFiles } from "./opencode/rewind.js"; +import type { ManagedProcessRegistry } from "../../managed-processes/managed-processes.js"; const OPENCODE_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: true, @@ -1214,6 +1215,7 @@ export const __openCodeInternals = { interface OpenCodeAgentClientDeps { runtime?: OpenCodeRuntime; + managedProcesses?: ManagedProcessRegistry; } class ProductionOpenCodeRuntime implements OpenCodeRuntime { @@ -1260,7 +1262,9 @@ export class OpenCodeAgentClient implements AgentClient { this.runtime = deps.runtime ?? new ProductionOpenCodeRuntime( - OpenCodeServerManager.getInstance(this.logger, runtimeSettings), + OpenCodeServerManager.getInstance(this.logger, runtimeSettings, { + managedProcesses: deps.managedProcesses, + }), ); } diff --git a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts index 867188233..7f06d6c47 100644 --- a/packages/server/src/server/agent/providers/opencode-server-manager.test.ts +++ b/packages/server/src/server/agent/providers/opencode-server-manager.test.ts @@ -1,42 +1,41 @@ +import type { ChildProcess } from "node:child_process"; import { EventEmitter } from "node:events"; -import { describe, expect, test, vi } from "vitest"; +import { describe, expect, test } from "vitest"; import { createTestLogger } from "../../../test-utils/test-logger.js"; -import { OpenCodeServerManager, type OpenCodeServerGeneration } from "./opencode/server-manager.js"; - -type FakeServerProcess = EventEmitter & { - killed: boolean; - kill: ReturnType; -}; - -type FakeGeneration = OpenCodeServerGeneration & { process: FakeServerProcess }; +import type { + ManagedProcessRecord, + ManagedProcessRecordInput, + ManagedProcessRegistry, + ManagedProcessReapResult, +} from "../../managed-processes/managed-processes.js"; +import type { ProcessTerminator, TreeKillTarget } from "../../../utils/tree-kill.js"; +import { + OpenCodeServerManager, + type OpenCodeCommandPrefixResolver, + type OpenCodePortAllocator, + type OpenCodeServerProcessSpawner, +} from "./opencode/server-manager.js"; describe("OpenCodeServerManager generations", () => { test("rotation creates a new current server without killing a referenced old server", async () => { - const manager = createTestManager(); - const first = createGeneration(4101); - const second = createGeneration(4102); - stubGenerations(manager, [first, second]); + const { manager, runtime } = createTestManager([4101, 4102]); const oldAcquisition = await manager.acquire({ force: false }); const newAcquisition = await manager.acquire({ force: true }); expect(oldAcquisition.server.url).toBe("http://127.0.0.1:4101"); expect(newAcquisition.server.url).toBe("http://127.0.0.1:4102"); - expect(first.process.kill).not.toHaveBeenCalled(); - expect(second.process.kill).not.toHaveBeenCalled(); + expect(runtime.terminatedPorts).toEqual([]); newAcquisition.release(); oldAcquisition.release(); - expect(first.process.kill).toHaveBeenCalledWith("SIGTERM"); + expect(runtime.terminatedPorts).toEqual([4101]); }); test("new acquisitions after rotation use the new server", async () => { - const manager = createTestManager(); - const first = createGeneration(4201); - const second = createGeneration(4202); - stubGenerations(manager, [first, second]); + const { manager, runtime } = createTestManager([4201, 4202]); const oldAcquisition = await manager.acquire({ force: false }); const rotatedAcquisition = await manager.acquire({ force: true }); @@ -45,18 +44,14 @@ describe("OpenCodeServerManager generations", () => { const nextAcquisition = await manager.acquire({ force: false }); expect(nextAcquisition.server.url).toBe("http://127.0.0.1:4202"); - expect(first.process.kill).not.toHaveBeenCalled(); + expect(runtime.terminatedPorts).toEqual([]); nextAcquisition.release(); oldAcquisition.release(); }); test("concurrent forced acquisitions share one fresh generation", async () => { - const manager = createTestManager(); - const first = createGeneration(4251); - const second = createGeneration(4252); - const third = createGeneration(4253); - const startServer = stubGenerations(manager, [first, second, third]); + const { manager, runtime } = createTestManager([4251, 4252, 4253]); const initialAcquisition = await manager.acquire({ force: false }); initialAcquisition.release(); @@ -68,17 +63,14 @@ describe("OpenCodeServerManager generations", () => { expect(modelsAcquisition.server.url).toBe("http://127.0.0.1:4252"); expect(modesAcquisition.server.url).toBe("http://127.0.0.1:4252"); - expect(startServer).toHaveBeenCalledTimes(2); + expect(runtime.launchedPorts).toEqual([4251, 4252]); modesAcquisition.release(); modelsAcquisition.release(); }); test("release is idempotent", async () => { - const manager = createTestManager(); - const first = createGeneration(4301); - const second = createGeneration(4302); - stubGenerations(manager, [first, second]); + const { manager, runtime } = createTestManager([4301, 4302]); const oldAcquisition = await manager.acquire({ force: false }); const newAcquisition = await manager.acquire({ force: true }); @@ -87,44 +79,33 @@ describe("OpenCodeServerManager generations", () => { oldAcquisition.release(); oldAcquisition.release(); - expect(first.refCount).toBe(0); - expect(first.process.kill).toHaveBeenCalledTimes(1); + expect(runtime.terminatedPorts).toEqual([4301]); }); test("shutdown kills current and retired servers", async () => { - const manager = createTestManager(); - const first = createGeneration(4401); - const second = createGeneration(4402); - stubGenerations(manager, [first, second]); + const { manager, runtime } = createTestManager([4401, 4402]); await manager.acquire({ force: false }); await manager.acquire({ force: true }); await manager.shutdown(); - expect(first.process.kill).toHaveBeenCalledWith("SIGTERM"); - expect(second.process.kill).toHaveBeenCalledWith("SIGTERM"); + expect(runtime.terminatedPorts).toEqual([4402, 4401]); }); test("shutdown still signals a process after an earlier kill signal if it has not exited", async () => { - const manager = createTestManager(); - const first = createGeneration(4451); - stubGenerations(manager, [first]); + const { manager, runtime } = createTestManager([4451]); await manager.acquire({ force: false }); - first.process.killed = true; + runtime.processForPort(4451).markKillSignalSent(); await manager.shutdown(); - expect(first.process.kill).toHaveBeenCalledWith("SIGTERM"); + expect(runtime.terminatedPorts).toEqual([4451]); }); test("repeated rotations leave zero unreferenced retired servers", async () => { - const manager = createTestManager(); - const first = createGeneration(4501); - const second = createGeneration(4502); - const third = createGeneration(4503); - stubGenerations(manager, [first, second, third]); + const { manager, runtime } = createTestManager([4501, 4502, 4503]); const firstAcquisition = await manager.acquire({ force: false }); const secondAcquisition = await manager.acquire({ force: true }); @@ -133,49 +114,204 @@ describe("OpenCodeServerManager generations", () => { thirdAcquisition.release(); firstAcquisition.release(); - const retiredServers = (manager as unknown as { retiredServers: Set }) - .retiredServers; - expect(Array.from(retiredServers).filter((server) => server.refCount === 0)).toHaveLength(0); - expect(first.process.kill).toHaveBeenCalledTimes(1); - expect(second.process.kill).toHaveBeenCalledTimes(1); + expect(runtime.terminatedPorts).toEqual([4502, 4501]); }); }); -function createTestManager(): OpenCodeServerManager { - const ManagerConstructor = OpenCodeServerManager as unknown as { - new (logger: ReturnType): OpenCodeServerManager; - }; - return new ManagerConstructor(createTestLogger()); -} +describe("OpenCodeServerManager managed process ledger", () => { + test("records helper server starts and removes the record on process exit", async () => { + const { manager, runtime } = createTestManager([4601]); -function stubGenerations( - manager: OpenCodeServerManager, - generations: FakeGeneration[], -): ReturnType { - const startServer = vi.fn(async () => { - const generation = generations.shift(); - if (!generation) { - throw new Error("No fake OpenCode server generation available"); - } - return generation; - }); - (manager as unknown as { startServer: typeof startServer }).startServer = startServer; - return startServer; -} + await manager.acquire({ force: false }); -function createGeneration(port: number): FakeGeneration { - const process = new EventEmitter() as FakeServerProcess; - process.killed = false; - process.kill = vi.fn((signal?: NodeJS.Signals) => { - process.killed = true; - process.emit("exit", signal ?? "SIGTERM"); - return true; + expect(await runtime.managedProcesses.list()).toEqual([ + { + id: "managed-process-1", + owner: { provider: "opencode", kind: "helper-server" }, + pid: 14601, + command: "opencode", + args: ["serve", "--port", "4601"], + metadata: { port: 4601 }, + identity: { commandLine: null, startedAt: null }, + createdAt: "test-created-at", + }, + ]); + + runtime.processForPort(4601).exitNormally(); + await runtime.settle(); + + expect(await runtime.managedProcesses.list()).toEqual([]); }); + + test("removes helper server records on shutdown", async () => { + const { manager, runtime } = createTestManager([4602]); + + await manager.acquire({ force: false }); + + await manager.shutdown(); + + expect(runtime.terminatedPorts).toEqual([4602]); + expect(await runtime.managedProcesses.list()).toEqual([]); + }); +}); + +function createTestManager(ports: number[]): { + manager: OpenCodeServerManager; + runtime: FakeOpenCodeServerRuntime; +} { + const runtime = new FakeOpenCodeServerRuntime(ports); return { - process, - port, - url: `http://127.0.0.1:${port}`, - refCount: 0, - retired: false, + manager: new OpenCodeServerManager({ + logger: createTestLogger(), + managedProcesses: runtime.managedProcesses, + portAllocator: runtime.allocatePort, + resolveCommandPrefix: runtime.resolveCommandPrefix, + spawnServerProcess: runtime.spawnServerProcess, + terminateProcess: runtime.terminateProcess, + }), + runtime, }; } + +class FakeOpenCodeServerRuntime { + readonly managedProcesses = new FakeManagedProcesses(); + readonly terminatedPorts: number[] = []; + private readonly ports: number[]; + private readonly processesByChild = new Map(); + private readonly processesByPort = new Map(); + + constructor(ports: number[]) { + this.ports = [...ports]; + } + + get launchedPorts(): number[] { + return Array.from(this.processesByPort.keys()); + } + + readonly allocatePort: OpenCodePortAllocator = async () => { + const port = this.ports.shift(); + if (!port) { + throw new Error("No fake OpenCode port available"); + } + return port; + }; + + readonly resolveCommandPrefix: OpenCodeCommandPrefixResolver = async () => ({ + command: "opencode", + args: [], + }); + + readonly spawnServerProcess: OpenCodeServerProcessSpawner = (command, args) => { + const port = Number(args.at(-1)); + const process = new FakeOpenCodeProcess({ port, pid: 10_000 + port }); + this.processesByChild.set(process.child, process); + this.processesByPort.set(port, process); + queueMicrotask(() => process.announceListening()); + return process.child; + }; + + readonly terminateProcess: ProcessTerminator = async (target: TreeKillTarget) => { + const process = this.processForChild(target as ChildProcess); + this.terminatedPorts.push(process.port); + process.exitBySignal("SIGTERM"); + return "terminated"; + }; + + processForPort(port: number): FakeOpenCodeProcess { + const process = this.processesByPort.get(port); + if (!process) { + throw new Error(`No fake OpenCode process for port ${port}`); + } + return process; + } + + async settle(): Promise { + await Promise.resolve(); + await Promise.resolve(); + } + + private processForChild(child: ChildProcess): FakeOpenCodeProcess { + const process = this.processesByChild.get(child); + if (!process) { + throw new Error("Unknown fake OpenCode process"); + } + return process; + } +} + +class FakeOpenCodeProcess extends EventEmitter { + readonly stdout = new EventEmitter(); + readonly stderr = new EventEmitter(); + readonly child: ChildProcess; + readonly port: number; + readonly pid: number; + killed = false; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + + constructor(options: { port: number; pid: number }) { + super(); + this.port = options.port; + this.pid = options.pid; + this.child = this as unknown as ChildProcess; + } + + announceListening(): void { + this.stdout.emit("data", Buffer.from("listening on")); + } + + exitNormally(): void { + this.exitCode = 0; + this.emit("exit", 0, null); + } + + exitBySignal(signal: NodeJS.Signals): void { + this.killed = true; + this.signalCode = signal; + this.emit("exit", null, signal); + } + + markKillSignalSent(): void { + this.killed = true; + } + + kill(signal?: NodeJS.Signals): boolean { + this.exitBySignal(signal ?? "SIGTERM"); + return true; + } +} + +class FakeManagedProcesses implements ManagedProcessRegistry { + private records: ManagedProcessRecord[] = []; + + async record(input: ManagedProcessRecordInput): Promise { + const record: ManagedProcessRecord = { + id: `managed-process-${this.records.length + 1}`, + ...input, + metadata: input.metadata ?? {}, + identity: { commandLine: null, startedAt: null }, + createdAt: "test-created-at", + }; + this.records.push(record); + return record; + } + + async remove(id: string): Promise { + this.records = this.records.filter((record) => record.id !== id); + } + + async list(): Promise { + return this.records; + } + + async reapStale(): Promise { + return { + checked: 0, + dead: 0, + mismatched: 0, + removed: 0, + terminated: 0, + errors: [], + }; + } +} diff --git a/packages/server/src/server/agent/providers/opencode/server-manager.ts b/packages/server/src/server/agent/providers/opencode/server-manager.ts index e57dd459a..04c7e6ff5 100644 --- a/packages/server/src/server/agent/providers/opencode/server-manager.ts +++ b/packages/server/src/server/agent/providers/opencode/server-manager.ts @@ -4,8 +4,9 @@ import os from "node:os"; import type { Logger } from "pino"; import { findExecutable } from "../../../../executable-resolution/executable-resolution.js"; -import { spawnProcess } from "../../../../utils/spawn.js"; -import { terminateWithTreeKill } from "../../../../utils/tree-kill.js"; +import { spawnProcess, type SpawnProcessOptions } from "../../../../utils/spawn.js"; +import { terminateWithTreeKill, type ProcessTerminator } from "../../../../utils/tree-kill.js"; +import type { ManagedProcessRegistry } from "../../../managed-processes/managed-processes.js"; import { createProviderEnvSpec, resolveProviderCommandPrefix, @@ -34,6 +35,25 @@ export interface OpenCodeServerGeneration { url: string; refCount: number; retired: boolean; + managedProcessId?: string; +} + +export type OpenCodePortAllocator = () => Promise; +export type OpenCodeCommandPrefixResolver = () => Promise<{ command: string; args: string[] }>; +export type OpenCodeServerProcessSpawner = ( + command: string, + args: string[], + options: SpawnProcessOptions, +) => ChildProcess; + +export interface OpenCodeServerManagerOptions { + logger: Logger; + runtimeSettings?: ProviderRuntimeSettings; + managedProcesses?: ManagedProcessRegistry; + terminateProcess?: ProcessTerminator; + portAllocator?: OpenCodePortAllocator; + resolveCommandPrefix?: OpenCodeCommandPrefixResolver; + spawnServerProcess?: OpenCodeServerProcessSpawner; } export class OpenCodeServerManager implements OpenCodeServerManagerLike { @@ -46,20 +66,37 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { private readonly logger: Logger; private readonly runtimeSettings?: ProviderRuntimeSettings; private readonly runtimeSettingsKey: string; + private readonly managedProcesses?: ManagedProcessRegistry; + private readonly terminateProcess: ProcessTerminator; + private readonly portAllocator: OpenCodePortAllocator; + private readonly resolveCommandPrefix: OpenCodeCommandPrefixResolver; + private readonly spawnServerProcess: OpenCodeServerProcessSpawner; - private constructor(logger: Logger, runtimeSettings?: ProviderRuntimeSettings) { - this.logger = logger; - this.runtimeSettings = runtimeSettings; - this.runtimeSettingsKey = JSON.stringify(runtimeSettings ?? {}); + constructor(options: OpenCodeServerManagerOptions) { + this.logger = options.logger; + this.runtimeSettings = options.runtimeSettings; + this.runtimeSettingsKey = JSON.stringify(this.runtimeSettings ?? {}); + this.managedProcesses = options.managedProcesses; + this.terminateProcess = options.terminateProcess ?? terminateWithTreeKill; + this.portAllocator = options.portAllocator ?? findAvailablePort; + this.resolveCommandPrefix = + options.resolveCommandPrefix ?? + (() => resolveProviderCommandPrefix(this.runtimeSettings?.command, resolveOpenCodeBinary)); + this.spawnServerProcess = options.spawnServerProcess ?? spawnProcess; } static getInstance( logger: Logger, runtimeSettings?: ProviderRuntimeSettings, + options: Omit = {}, ): OpenCodeServerManager { const nextSettingsKey = JSON.stringify(runtimeSettings ?? {}); if (!OpenCodeServerManager.instance) { - OpenCodeServerManager.instance = new OpenCodeServerManager(logger, runtimeSettings); + OpenCodeServerManager.instance = new OpenCodeServerManager({ + logger, + runtimeSettings, + ...options, + }); OpenCodeServerManager.registerExitHandler(); } else if (OpenCodeServerManager.instance.runtimeSettingsKey !== nextSettingsKey) { logger.warn( @@ -190,27 +227,28 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { } private async startServer(launchEnv?: Record): Promise { - const port = await findAvailablePort(); + const port = await this.portAllocator(); const url = `http://127.0.0.1:${port}`; - const launchPrefix = await resolveProviderCommandPrefix( - this.runtimeSettings?.command, - resolveOpenCodeBinary, - ); + const launchPrefix = await this.resolveCommandPrefix(); + const serverArgs = [...launchPrefix.args, "serve", "--port", String(port)]; + const serverCwd = os.homedir(); return new Promise((resolve, reject) => { - const serverProcess = spawnProcess( - launchPrefix.command, - [...launchPrefix.args, "serve", "--port", String(port)], - { - cwd: os.homedir(), - detached: process.platform !== "win32", - stdio: ["ignore", "pipe", "pipe"], - ...createProviderEnvSpec({ - runtimeSettings: this.runtimeSettings, - overlays: [launchEnv], - }), - }, - ); + const serverProcess = this.spawnServerProcess(launchPrefix.command, serverArgs, { + cwd: serverCwd, + detached: process.platform !== "win32", + stdio: ["ignore", "pipe", "pipe"], + ...createProviderEnvSpec({ + runtimeSettings: this.runtimeSettings, + overlays: [launchEnv], + }), + }); + const managedProcessRecord = this.recordManagedServerProcess({ + process: serverProcess, + command: launchPrefix.command, + args: serverArgs, + port, + }); let started = false; let stderrBuffer = ""; @@ -247,13 +285,17 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { if (output.includes("listening on") && !started) { started = true; clearTimeout(timeout); - resolve({ - process: serverProcess, - port, - url, - refCount: 0, - retired: false, - }); + void (async () => { + const record = await managedProcessRecord; + resolve({ + process: serverProcess, + port, + url, + refCount: 0, + retired: false, + ...(record ? { managedProcessId: record.id } : {}), + }); + })(); } }); @@ -265,11 +307,13 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { serverProcess.on("error", (error) => { clearTimeout(timeout); + this.removeManagedProcessRecordWhenResolved(managedProcessRecord); const headline = error instanceof Error ? error.message : String(error); reject(new Error(buildStartupErrorMessage(headline))); }); serverProcess.on("exit", (code) => { + this.removeManagedProcessRecordWhenResolved(managedProcessRecord); if (!started) { clearTimeout(timeout); reject(new Error(buildStartupErrorMessage(`OpenCode server exited with code ${code}`))); @@ -312,7 +356,7 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { ) { return; } - const result = await terminateWithTreeKill(server.process, { + const result = await this.terminateProcess(server.process, { gracefulTimeoutMs: OPENCODE_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS, forceTimeoutMs: OPENCODE_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS, onForceSignal: () => { @@ -328,6 +372,55 @@ export class OpenCodeServerManager implements OpenCodeServerManagerLike { "OpenCode server did not report exit after SIGKILL", ); } + if (server.managedProcessId) { + await this.removeManagedProcessId(server.managedProcessId); + server.managedProcessId = undefined; + } + } + + private async recordManagedServerProcess(options: { + process: ChildProcess; + command: string; + args: string[]; + port: number; + }): Promise<{ id: string } | null> { + const pid = options.process.pid; + if (!this.managedProcesses || typeof pid !== "number" || pid <= 0) { + return null; + } + + try { + return await this.managedProcesses.record({ + owner: { provider: "opencode", kind: "helper-server" }, + pid, + command: options.command, + args: options.args, + metadata: { port: options.port }, + }); + } catch (error) { + this.logger.warn( + { err: error, pid, port: options.port }, + "Failed to record OpenCode helper process", + ); + return null; + } + } + + private removeManagedProcessRecordWhenResolved(record: Promise<{ id: string } | null>): void { + void record.then((resolved) => { + if (resolved) { + return this.removeManagedProcessId(resolved.id); + } + return undefined; + }); + } + + private async removeManagedProcessId(id: string): Promise { + try { + await this.managedProcesses?.remove(id); + } catch (error) { + this.logger.warn({ err: error, id }, "Failed to remove OpenCode helper process record"); + } } } diff --git a/packages/server/src/server/bootstrap-managed-processes.test.ts b/packages/server/src/server/bootstrap-managed-processes.test.ts new file mode 100644 index 000000000..d78cabd75 --- /dev/null +++ b/packages/server/src/server/bootstrap-managed-processes.test.ts @@ -0,0 +1,90 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import pino from "pino"; +import { afterEach, describe, expect, test } from "vitest"; + +import type { + ManagedProcessRecord, + ManagedProcessRecordInput, + ManagedProcessRegistry, + ManagedProcessReapResult, +} from "./managed-processes/managed-processes.js"; +import { createPaseoDaemon, type PaseoDaemonConfig } from "./bootstrap.js"; +import { createTestAgentClients } from "./test-utils/fake-agent-client.js"; + +let tempRoot: string | null = null; +let staticDir: string | null = null; + +afterEach(async () => { + await Promise.all([ + tempRoot ? rm(tempRoot, { recursive: true, force: true }) : Promise.resolve(), + staticDir ? rm(staticDir, { recursive: true, force: true }) : Promise.resolve(), + ]); + tempRoot = null; + staticDir = null; +}); + +describe("daemon managed process bootstrap", () => { + test("reaps stale helper process records during daemon bootstrap", async () => { + tempRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-managed-bootstrap-")); + staticDir = await mkdtemp(path.join(os.tmpdir(), "paseo-static-")); + const paseoHome = path.join(tempRoot, ".paseo"); + const managedProcesses = new FakeManagedProcesses(); + const daemon = await createPaseoDaemon( + { + listen: "127.0.0.1:0", + paseoHome, + corsAllowedOrigins: [], + hostnames: true, + mcpEnabled: false, + staticDir, + mcpDebug: false, + agentClients: createTestAgentClients(), + agentStoragePath: path.join(paseoHome, "agents"), + relayEnabled: false, + appBaseUrl: "https://app.paseo.sh", + managedProcesses, + } as PaseoDaemonConfig, + pino({ level: "silent" }), + ); + + try { + expect(managedProcesses.reapCount).toBe(1); + } finally { + await daemon.stop().catch(() => undefined); + } + }); +}); + +class FakeManagedProcesses implements ManagedProcessRegistry { + reapCount = 0; + + async record(input: ManagedProcessRecordInput): Promise { + return { + id: "unused", + ...input, + metadata: input.metadata ?? {}, + identity: { commandLine: null, startedAt: null }, + createdAt: "unused", + }; + } + + async remove(): Promise {} + + async list(): Promise { + return []; + } + + async reapStale(): Promise { + this.reapCount += 1; + return { + checked: 1, + dead: 0, + mismatched: 0, + removed: 1, + terminated: 1, + errors: [], + }; + } +} diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 703ac337a..1dac2d2da 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -142,6 +142,12 @@ import { createServiceProxySubsystem, type ServiceProxySubsystem } from "./servi import { ScriptHealthMonitor } from "./script-health-monitor.js"; import { createScriptStatusEmitter } from "./script-status-projection.js"; import { WorkspaceScriptRuntimeStore } from "./workspace-script-runtime-store.js"; +import { + createManagedProcessRegistry, + createSystemManagedProcessTable, + type ManagedProcessRegistry, +} from "./managed-processes/managed-processes.js"; +import { terminateWithTreeKill } from "../utils/tree-kill.js"; import { isHostnameAllowed, type HostnamesConfig } from "./hostnames.js"; import { createRequireBearerMiddleware, @@ -355,6 +361,7 @@ export interface PaseoDaemonConfig { log?: PersistedConfig["log"]; onLifecycleIntent?: (intent: DaemonLifecycleIntent) => void; pushNotificationSender?: PushNotificationSender; + managedProcesses?: ManagedProcessRegistry; } export interface PaseoDaemon { @@ -369,6 +376,32 @@ export interface PaseoDaemon { getListenTarget(): ListenTarget | null; } +function createBootstrapManagedProcessRegistry( + config: Pick, + logger: Logger, +): ManagedProcessRegistry { + if (config.managedProcesses) { + return config.managedProcesses; + } + + return createManagedProcessRegistry({ + paseoHome: config.paseoHome, + processTable: createSystemManagedProcessTable(), + terminateProcess: terminateWithTreeKill, + logger, + }); +} + +async function reconcileManagedProcessLedger( + managedProcesses: ManagedProcessRegistry, + logger: Logger, +): Promise { + const reapResult = await managedProcesses.reapStale(); + if (reapResult.checked > 0 || reapResult.errors.length > 0) { + logger.info(reapResult, "Managed helper process ledger reconciled"); + } +} + export async function createPaseoDaemon( config: PaseoDaemonConfig, rootLogger: Logger, @@ -405,6 +438,13 @@ export async function createPaseoDaemon( const serverId = getOrCreateServerId(config.paseoHome, { logger }); const daemonKeyPair = await loadOrCreateDaemonKeyPair(config.paseoHome, logger); + const managedProcesses = createBootstrapManagedProcessRegistry(config, logger); + // Reconcile the helper-process ledger in the background so it never blocks the + // daemon from coming up; terminating a live leftover can take a few seconds. + // Best-effort, so a failure is logged here rather than crashing startup. + void reconcileManagedProcessLedger(managedProcesses, logger).catch((error) => { + logger.warn({ err: error }, "Failed to reconcile managed helper process ledger"); + }); let relayTransport: RelayTransportController | null = null; const staticDir = config.staticDir; @@ -645,6 +685,7 @@ export async function createPaseoDaemon( runtimeSettings: config.agentProviderSettings, providerOverrides: config.providerOverrides, workspaceGitService, + managedProcesses, isDev: config.isDev === true, extraClients: config.agentClients, }); diff --git a/packages/server/src/server/managed-processes/managed-processes.test.ts b/packages/server/src/server/managed-processes/managed-processes.test.ts new file mode 100644 index 000000000..e247db027 --- /dev/null +++ b/packages/server/src/server/managed-processes/managed-processes.test.ts @@ -0,0 +1,391 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, test } from "vitest"; + +import { createTestLogger } from "../../test-utils/test-logger.js"; +import { + createManagedProcessRegistry, + createPidTarget, + createSystemManagedProcessTable, + type ManagedProcessCommandRunner, + type ManagedProcessInspection, + type ManagedProcessSnapshot, + type ManagedProcessTable, +} from "./managed-processes.js"; +import { spawnProcess } from "../../utils/spawn.js"; +import { + terminateWithTreeKill, + type ProcessTerminator, + type TreeKillTarget, +} from "../../utils/tree-kill.js"; + +let tempHome: string | null = null; + +afterEach(async () => { + if (tempHome) { + await rm(tempHome, { recursive: true, force: true }); + tempHome = null; + } +}); + +describe("managed process registry", () => { + test("reaps a validated leftover helper process and deletes its record", async () => { + tempHome = await mkdtemp(path.join(tmpdir(), "paseo-managed-processes-")); + const processTable = new FakeProcessTable([ + { + pid: 4101, + commandLine: "opencode serve --port 4101", + startedAt: "process-start-token", + }, + ]); + const terminator = new FakeProcessTerminator(); + const registry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable, + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + await registry.record({ + owner: { provider: "opencode", kind: "helper-server" }, + pid: 4101, + command: "opencode", + args: ["serve", "--port", "4101"], + metadata: { port: 4101 }, + }); + + const restartedRegistry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable, + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + const result = await restartedRegistry.reapStale(); + + expect(result).toEqual({ + checked: 1, + dead: 0, + mismatched: 0, + removed: 1, + terminated: 1, + errors: [], + }); + expect(terminator.terminatedPids).toEqual([4101]); + expect(await restartedRegistry.list()).toEqual([]); + }); + + test("deletes a dead helper process record without terminating a PID", async () => { + tempHome = await mkdtemp(path.join(tmpdir(), "paseo-managed-processes-")); + const processTable = new FakeProcessTable([ + { + pid: 4102, + commandLine: "opencode serve --port 4102", + startedAt: "process-start-token", + }, + ]); + const terminator = new FakeProcessTerminator(); + const registry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable, + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + await registry.record({ + owner: { provider: "opencode", kind: "helper-server" }, + pid: 4102, + command: "opencode", + args: ["serve", "--port", "4102"], + metadata: { port: 4102 }, + }); + + const restartedRegistry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable: new FakeProcessTable([]), + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + const result = await restartedRegistry.reapStale(); + + expect(result).toEqual({ + checked: 1, + dead: 1, + mismatched: 0, + removed: 1, + terminated: 0, + errors: [], + }); + expect(terminator.terminatedPids).toEqual([]); + expect(await restartedRegistry.list()).toEqual([]); + }); + + test("removes a reused PID record without terminating the new process", async () => { + tempHome = await mkdtemp(path.join(tmpdir(), "paseo-managed-processes-")); + const terminator = new FakeProcessTerminator(); + const registry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable: new FakeProcessTable([ + { + pid: 4103, + commandLine: "opencode serve --port 4103", + startedAt: "original-start-token", + }, + ]), + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + await registry.record({ + owner: { provider: "opencode", kind: "helper-server" }, + pid: 4103, + command: "opencode", + args: ["serve", "--port", "4103"], + metadata: { port: 4103 }, + }); + + const restartedRegistry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable: new FakeProcessTable([ + { + pid: 4103, + commandLine: "opencode serve --port 4103", + startedAt: "new-process-start-token", + }, + ]), + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + const result = await restartedRegistry.reapStale(); + + expect(result).toEqual({ + checked: 1, + dead: 0, + mismatched: 1, + removed: 1, + terminated: 0, + errors: [], + }); + expect(terminator.terminatedPids).toEqual([]); + expect(await restartedRegistry.list()).toEqual([]); + }); + + test("keeps a helper record when inspection fails instead of orphaning a live process", async () => { + tempHome = await mkdtemp(path.join(tmpdir(), "paseo-managed-processes-")); + const terminator = new FakeProcessTerminator(); + const registry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable: new FakeProcessTable([ + { pid: 4104, commandLine: "opencode serve --port 4104", startedAt: "process-start-token" }, + ]), + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + await registry.record({ + owner: { provider: "opencode", kind: "helper-server" }, + pid: 4104, + command: "opencode", + args: ["serve", "--port", "4104"], + metadata: { port: 4104 }, + }); + + const restartedRegistry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable: new FakeProcessTable([], [4104]), + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + const result = await restartedRegistry.reapStale(); + + expect(result).toMatchObject({ + checked: 1, + dead: 0, + mismatched: 0, + removed: 0, + terminated: 0, + }); + expect(result.errors).toEqual([{ id: expect.any(String), message: "inspection failed" }]); + expect(terminator.terminatedPids).toEqual([]); + expect(await restartedRegistry.list()).toHaveLength(1); + }); + + test("does not terminate a reused PID whose command line only mentions the tokens", async () => { + tempHome = await mkdtemp(path.join(tmpdir(), "paseo-managed-processes-")); + const terminator = new FakeProcessTerminator(); + const registry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable: new FakeProcessTable([], [4105]), + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + await registry.record({ + owner: { provider: "opencode", kind: "helper-server" }, + pid: 4105, + command: "opencode", + args: ["serve", "--port", "4105"], + metadata: { port: 4105 }, + }); + + const restartedRegistry = createManagedProcessRegistry({ + paseoHome: tempHome, + processTable: new FakeProcessTable([ + { + pid: 4105, + commandLine: "node /tmp/serve.js --port 4105 # opencode helper", + startedAt: null, + }, + ]), + terminateProcess: terminator.terminate, + logger: createTestLogger(), + }); + const result = await restartedRegistry.reapStale(); + + expect(result).toEqual({ + checked: 1, + dead: 0, + mismatched: 1, + removed: 1, + terminated: 0, + errors: [], + }); + expect(terminator.terminatedPids).toEqual([]); + expect(await restartedRegistry.list()).toEqual([]); + }); +}); + +describe("managed process termination", () => { + test("stops as soon as a terminated process exits instead of escalating to SIGKILL", async () => { + const child = spawnProcess(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { + stdio: "ignore", + }); + const pid = child.pid; + if (!pid) { + throw new Error("Failed to spawn test process"); + } + + let forced = false; + const result = await terminateWithTreeKill(createPidTarget(pid), { + gracefulTimeoutMs: 2_000, + forceTimeoutMs: 1_000, + onForceSignal: () => { + forced = true; + }, + }); + + expect(result).toBe("terminated"); + expect(forced).toBe(false); + }); +}); + +describe("system managed process table", () => { + test("reads POSIX process identity from ps", async () => { + const commandRunner = new FakeCommandRunner([ + { + stdout: "Sat Jun 20 10:30:40 2026 opencode serve --port 4101\n", + stderr: "", + }, + ]); + const processTable = createSystemManagedProcessTable({ + platform: "darwin", + commandRunner, + }); + + const inspection = await processTable.inspect(4101); + + expect(inspection).toEqual({ + status: "alive", + snapshot: { + pid: 4101, + commandLine: "opencode serve --port 4101", + startedAt: "Sat Jun 20 10:30:40 2026", + }, + }); + expect(commandRunner.commands).toEqual([ + { + command: "ps", + args: ["-ww", "-p", "4101", "-o", "lstart=", "-o", "command="], + }, + ]); + }); + + test("reads Windows process identity from PowerShell", async () => { + const commandRunner = new FakeCommandRunner([ + { + stdout: JSON.stringify({ + ProcessId: 4101, + CommandLine: "C:\\opencode.exe serve --port 4101", + CreationDate: "20260620103040.000000+000", + }), + stderr: "", + }, + ]); + const processTable = createSystemManagedProcessTable({ + platform: "win32", + commandRunner, + }); + + const inspection = await processTable.inspect(4101); + + expect(inspection).toEqual({ + status: "alive", + snapshot: { + pid: 4101, + commandLine: "C:\\opencode.exe serve --port 4101", + startedAt: "20260620103040.000000+000", + }, + }); + expect(commandRunner.commands).toEqual([ + { + command: "powershell.exe", + args: [ + "-NoProfile", + "-NonInteractive", + "-Command", + "$process = Get-CimInstance Win32_Process -Filter 'ProcessId = 4101'; if ($process) { $process | Select-Object ProcessId,CommandLine,CreationDate | ConvertTo-Json -Compress }", + ], + }, + ]); + }); +}); + +class FakeProcessTable implements ManagedProcessTable { + private readonly snapshots: Map; + private readonly errorPids: Set; + + constructor(snapshots: ManagedProcessSnapshot[], errorPids: number[] = []) { + this.snapshots = new Map(snapshots.map((snapshot) => [snapshot.pid, snapshot])); + this.errorPids = new Set(errorPids); + } + + async inspect(pid: number): Promise { + if (this.errorPids.has(pid)) { + return { status: "error", error: new Error("inspection failed") }; + } + const snapshot = this.snapshots.get(pid); + return snapshot ? { status: "alive", snapshot } : { status: "not-found" }; + } +} + +class FakeProcessTerminator { + readonly terminatedPids: number[] = []; + + readonly terminate: ProcessTerminator = async (target: TreeKillTarget) => { + this.terminatedPids.push(target.pid ?? -1); + return "terminated"; + }; +} + +class FakeCommandRunner implements ManagedProcessCommandRunner { + readonly commands: Array<{ command: string; args: string[] }> = []; + private readonly responses: Array<{ stdout: string; stderr: string }>; + + constructor(responses: Array<{ stdout: string; stderr: string }>) { + this.responses = [...responses]; + } + + async exec(command: string, args: string[]): Promise<{ stdout: string; stderr: string }> { + this.commands.push({ command, args }); + const response = this.responses.shift(); + if (!response) { + throw new Error("No fake process-table command response available"); + } + return response; + } +} diff --git a/packages/server/src/server/managed-processes/managed-processes.ts b/packages/server/src/server/managed-processes/managed-processes.ts new file mode 100644 index 000000000..0031080c6 --- /dev/null +++ b/packages/server/src/server/managed-processes/managed-processes.ts @@ -0,0 +1,455 @@ +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import path from "node:path"; +import type { Logger } from "pino"; +import { z } from "zod"; +import { writeJsonFileAtomic } from "../atomic-file.js"; +import { execCommand } from "../../utils/spawn.js"; +import type { ProcessTerminator, TreeKillTarget } from "../../utils/tree-kill.js"; + +const MANAGED_PROCESS_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 5_000; +const MANAGED_PROCESS_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; +const MANAGED_PROCESS_EXIT_POLL_INTERVAL_MS = 50; +const MANAGED_PROCESS_ID_PATTERN = /^[a-zA-Z0-9_-]+$/; +// `ps -o lstart` emits a fixed-width 24-char ctime stamp, e.g. "Sat Jun 20 10:30:40 2026". +const POSIX_LSTART_WIDTH = 24; + +const ManagedProcessRecordSchema = z.object({ + id: z.string().min(1), + owner: z.object({ + provider: z.string().min(1), + kind: z.string().min(1), + }), + pid: z.number().int().positive(), + command: z.string().min(1), + args: z.array(z.string()), + metadata: z.record(z.string(), z.unknown()).default({}), + identity: z.object({ + commandLine: z.string().nullable(), + startedAt: z.string().nullable(), + }), + createdAt: z.string().min(1), +}); + +const WindowsProcessSnapshotSchema = z.object({ + ProcessId: z.number().int().positive(), + CommandLine: z.string().nullable().optional(), + CreationDate: z.string().nullable().optional(), +}); + +export interface ManagedProcessSnapshot { + pid: number; + commandLine: string | null; + startedAt: string | null; +} + +export type ManagedProcessInspection = + | { status: "alive"; snapshot: ManagedProcessSnapshot } + | { status: "not-found" } + | { status: "error"; error: unknown }; + +export interface ManagedProcessTable { + inspect(pid: number): Promise; +} + +export interface ManagedProcessCommandRunner { + exec(command: string, args: string[]): Promise<{ stdout: string; stderr: string }>; +} + +export interface ManagedProcessOwner { + provider: string; + kind: string; +} + +export interface ManagedProcessRecordInput { + owner: ManagedProcessOwner; + pid: number; + command: string; + args: string[]; + metadata?: Record; +} + +export interface ManagedProcessRecord extends ManagedProcessRecordInput { + id: string; + metadata: Record; + identity: { + commandLine: string | null; + startedAt: string | null; + }; + createdAt: string; +} + +export interface ManagedProcessReapResult { + checked: number; + dead: number; + mismatched: number; + removed: number; + terminated: number; + errors: Array<{ id: string; message: string }>; +} + +export interface ManagedProcessRegistry { + record(input: ManagedProcessRecordInput): Promise; + remove(id: string): Promise; + list(): Promise; + reapStale(): Promise; +} + +interface ManagedProcessRegistryOptions { + paseoHome: string; + processTable: ManagedProcessTable; + terminateProcess: ProcessTerminator; + logger: Logger; +} + +export function createManagedProcessRegistry( + options: ManagedProcessRegistryOptions, +): ManagedProcessRegistry { + return new FileBackedManagedProcessRegistry(options); +} + +export function createSystemManagedProcessTable(options?: { + platform?: NodeJS.Platform; + commandRunner?: ManagedProcessCommandRunner; +}): ManagedProcessTable { + return new SystemManagedProcessTable({ + platform: options?.platform ?? process.platform, + commandRunner: options?.commandRunner ?? { + exec: execCommand, + }, + }); +} + +class SystemManagedProcessTable implements ManagedProcessTable { + private readonly platform: NodeJS.Platform; + private readonly commandRunner: ManagedProcessCommandRunner; + + constructor(options: { platform: NodeJS.Platform; commandRunner: ManagedProcessCommandRunner }) { + this.platform = options.platform; + this.commandRunner = options.commandRunner; + } + + async inspect(pid: number): Promise { + if (!Number.isInteger(pid) || pid <= 0) { + return { status: "not-found" }; + } + + try { + return this.platform === "win32" + ? await this.inspectWindows(pid) + : await this.inspectPosix(pid); + } catch (error) { + return { status: "error", error }; + } + } + + private async inspectPosix(pid: number): Promise { + let stdout: string; + try { + ({ stdout } = await this.commandRunner.exec("ps", [ + "-ww", + "-p", + String(pid), + "-o", + "lstart=", + "-o", + "command=", + ])); + } catch (error) { + // `ps -p ` exits non-zero when no process matches the pid; a numeric + // exit code means ps ran and found nothing, distinct from ps failing to run. + return isCommandExitFailure(error) ? { status: "not-found" } : { status: "error", error }; + } + + const line = stdout.trimEnd(); + if (!line) { + return { status: "not-found" }; + } + + const startedAt = line.slice(0, POSIX_LSTART_WIDTH).trim(); + const commandLine = line.slice(POSIX_LSTART_WIDTH).trim(); + return { + status: "alive", + snapshot: { + pid, + commandLine: commandLine || null, + startedAt: startedAt || null, + }, + }; + } + + private async inspectWindows(pid: number): Promise { + const command = [ + `$process = Get-CimInstance Win32_Process -Filter 'ProcessId = ${pid}';`, + "if ($process) { $process | Select-Object ProcessId,CommandLine,CreationDate | ConvertTo-Json -Compress }", + ].join(" "); + const { stdout } = await this.commandRunner.exec("powershell.exe", [ + "-NoProfile", + "-NonInteractive", + "-Command", + command, + ]); + const trimmed = stdout.trim(); + if (!trimmed) { + return { status: "not-found" }; + } + + const parsed = WindowsProcessSnapshotSchema.parse(JSON.parse(trimmed)); + return { + status: "alive", + snapshot: { + pid, + commandLine: parsed.CommandLine ?? null, + startedAt: parsed.CreationDate ?? null, + }, + }; + } +} + +class FileBackedManagedProcessRegistry implements ManagedProcessRegistry { + private readonly directory: string; + private readonly processTable: ManagedProcessTable; + private readonly terminateProcess: ProcessTerminator; + private readonly logger: Logger; + + constructor(options: ManagedProcessRegistryOptions) { + this.directory = path.join(options.paseoHome, "runtime", "managed-processes"); + this.processTable = options.processTable; + this.terminateProcess = options.terminateProcess; + this.logger = options.logger.child({ module: "managed-processes" }); + } + + async record(input: ManagedProcessRecordInput): Promise { + const inspection = await this.processTable.inspect(input.pid); + const snapshot = inspection.status === "alive" ? inspection.snapshot : null; + const record: ManagedProcessRecord = { + id: randomUUID(), + owner: input.owner, + pid: input.pid, + command: input.command, + args: input.args, + metadata: input.metadata ?? {}, + identity: { + commandLine: snapshot?.commandLine ?? null, + startedAt: snapshot?.startedAt ?? null, + }, + createdAt: new Date().toISOString(), + }; + + await writeJsonFileAtomic(this.recordPath(record.id), record); + return record; + } + + async remove(id: string): Promise { + await fs.rm(this.recordPath(id), { force: true }); + } + + async list(): Promise { + const entries = await this.readEntries(); + return entries.map((entry) => entry.record); + } + + async reapStale(): Promise { + const result: ManagedProcessReapResult = { + checked: 0, + dead: 0, + mismatched: 0, + removed: 0, + terminated: 0, + errors: [], + }; + + for (const entry of await this.readEntries()) { + result.checked += 1; + try { + const inspection = await this.processTable.inspect(entry.record.pid); + if (inspection.status === "not-found") { + await fs.rm(entry.path, { force: true }); + result.dead += 1; + result.removed += 1; + continue; + } + + if (inspection.status === "error") { + // Inspection failed, so we cannot tell whether the helper is still + // alive. Keep the record and retry on the next reconcile rather than + // orphaning a live process by deleting its record without killing it. + const message = + inspection.error instanceof Error ? inspection.error.message : String(inspection.error); + result.errors.push({ id: entry.record.id, message }); + this.logger.warn( + { + err: inspection.error, + id: entry.record.id, + pid: entry.record.pid, + owner: entry.record.owner, + }, + "Could not inspect managed helper process; leaving record for next reconcile", + ); + continue; + } + + const snapshot = inspection.snapshot; + if (!processIdentityMatches(entry.record, snapshot)) { + await fs.rm(entry.path, { force: true }); + result.mismatched += 1; + result.removed += 1; + continue; + } + + await this.terminateProcess(createPidTarget(entry.record.pid), { + gracefulTimeoutMs: MANAGED_PROCESS_GRACEFUL_SHUTDOWN_TIMEOUT_MS, + forceTimeoutMs: MANAGED_PROCESS_FORCE_SHUTDOWN_TIMEOUT_MS, + onForceSignal: () => { + this.logger.warn( + { + pid: entry.record.pid, + owner: entry.record.owner, + timeoutMs: MANAGED_PROCESS_GRACEFUL_SHUTDOWN_TIMEOUT_MS, + }, + "Managed helper process did not exit after SIGTERM; sending SIGKILL", + ); + }, + }); + await fs.rm(entry.path, { force: true }); + result.terminated += 1; + result.removed += 1; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + result.errors.push({ id: entry.record.id, message }); + this.logger.warn( + { err: error, id: entry.record.id, pid: entry.record.pid, owner: entry.record.owner }, + "Failed to reap managed helper process", + ); + } + } + + return result; + } + + private recordPath(id: string): string { + if (!MANAGED_PROCESS_ID_PATTERN.test(id)) { + throw new Error(`Invalid managed process record id: ${id}`); + } + return path.join(this.directory, `${id}.json`); + } + + private async readEntries(): Promise> { + let fileNames: string[]; + try { + fileNames = await fs.readdir(this.directory); + } catch (error) { + if (isNodeErrorWithCode(error, "ENOENT")) { + return []; + } + throw error; + } + + const entries: Array<{ path: string; record: ManagedProcessRecord }> = []; + for (const fileName of fileNames) { + if (!fileName.endsWith(".json")) { + continue; + } + const filePath = path.join(this.directory, fileName); + try { + const raw = await fs.readFile(filePath, "utf8"); + const parsed = ManagedProcessRecordSchema.parse(JSON.parse(raw)); + entries.push({ path: filePath, record: parsed }); + } catch (error) { + // A single corrupt or partially-written record must not abort the whole + // reconcile and leave every other leftover un-reaped. Skip it. + this.logger.warn( + { err: error, file: fileName }, + "Skipping unreadable managed process record", + ); + } + } + return entries; + } +} + +function processIdentityMatches( + record: ManagedProcessRecord, + snapshot: ManagedProcessSnapshot, +): boolean { + if (record.identity.startedAt && snapshot.startedAt) { + if (record.identity.startedAt !== snapshot.startedAt) { + return false; + } + return snapshot.commandLine ? commandLineMatchesRecord(record, snapshot.commandLine) : true; + } + + if (record.identity.commandLine && snapshot.commandLine) { + return ( + normalizeCommandLine(record.identity.commandLine) === + normalizeCommandLine(snapshot.commandLine) + ); + } + + return snapshot.commandLine ? commandLineMatchesRecord(record, snapshot.commandLine) : false; +} + +function commandLineMatchesRecord(record: ManagedProcessRecord, commandLine: string): boolean { + // Require the command name and args as one contiguous run, not scattered + // tokens. Without exact process identity (lstart), a reused PID whose command + // line merely mentions "opencode", "serve" and the port elsewhere must not be + // mistaken for our leftover and killed. + const normalized = normalizeCommandLine(commandLine); + const commandName = path.basename(record.command).toLowerCase(); + const signature = [commandName, ...record.args].map((token) => token.toLowerCase()).join(" "); + return normalized.includes(signature); +} + +function normalizeCommandLine(commandLine: string): string { + return commandLine.replace(/\s+/g, " ").trim().toLowerCase(); +} + +export function createPidTarget(pid: number): TreeKillTarget { + return { + pid, + exitCode: null, + signalCode: null, + kill(signal?: NodeJS.Signals | number) { + process.kill(pid, signal); + return true; + }, + // The reaper has no ChildProcess handle for a leftover from a previous + // daemon, so it observes exit by polling the pid. Without this, termination + // can never see a graceful SIGTERM exit and always waits out the full + // graceful+force window before escalating to SIGKILL. + once(_event, listener) { + const timer = setInterval(() => { + if (!isProcessAlive(pid)) { + clearInterval(timer); + listener(); + } + }, MANAGED_PROCESS_EXIT_POLL_INTERVAL_MS); + timer.unref(); + }, + }; +} + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return isNodeErrorWithCode(error, "EPERM"); + } +} + +function isCommandExitFailure(error: unknown): boolean { + // execFile rejects with a numeric `code` (the process exit status) when the + // command ran and exited non-zero; a string `code` (e.g. "ENOENT") means it + // never ran. + return typeof (error as { code?: unknown })?.code === "number"; +} + +function isNodeErrorWithCode(error: unknown, code: string): boolean { + return ( + typeof error === "object" && + error !== null && + "code" in error && + (error as { code?: unknown }).code === code + ); +}