diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index d3c5767c03b..63fdb03c77b 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1297,6 +1297,50 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("terminates provider-group descendants after the guardian exits", async () => { + const guardian = spawn( + "/bin/sh", + ["-c", "sleep 3600 & echo $!; sleep 0.2"], + { detached: true, stdio: ["ignore", "pipe", "ignore"] }, + ); + childProcesses.add(guardian); + expect(guardian.pid).toBeTypeOf("number"); + const guardianStartedAt = await readProcessStartedAt(guardian.pid ?? 0); + const descendantPid = await new Promise((resolvePid, rejectPid) => { + guardian.stdout?.once("data", (chunk: Buffer) => { + const parsedPid = Number.parseInt(chunk.toString("utf8").trim(), 10); + if (Number.isInteger(parsedPid) && parsedPid > 0) resolvePid(parsedPid); + else rejectPid(new Error("Provider descendant pid was not emitted")); + }); + guardian.once("error", rejectPid); + }); + cleanupPids.add(descendantPid); + await new Promise((resolveExit) => guardian.once("exit", () => resolveExit())); + expect(isPidAlive(guardian.pid)).toBe(false); + expect(isPidAlive(descendantPid)).toBe(true); + + const { runId } = await seedRunFixture({ + processPid: 999_999_999, + providerProcessPid: guardian.pid ?? null, + providerProcessGroupId: guardian.pid ?? null, + providerProcessStartedAt: new Date(guardianStartedAt ?? 0), + includeIssue: false, + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.reapOrphanedRuns(); + + expect(result.reaped).toBe(1); + expect(await waitForPidExit(descendantPid)).toBe(true); + const run = await heartbeat.getRun(runId); + expect(run).toMatchObject({ + status: "failed", + providerProcessPid: null, + providerProcessGroupId: null, + providerProcessStartedAt: null, + }); + }); + it("skips generic timer wakes without invoking an adapter when no assigned work is actionable", async () => { const { companyId, agentId } = await seedIdleTimerAgentFixture(); const heartbeat = heartbeatService(db); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index fa90013938b..a07d5667804 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -9947,12 +9947,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) ): Promise { const pid = run.providerProcessPid; const processGroupId = run.providerProcessGroupId; - const alive = isProcessAlive(pid) || isProcessGroupAlive(processGroupId); + const pidAlive = isProcessAlive(pid); + const processGroupAlive = isProcessGroupAlive(processGroupId); let ownsLiveProcess = false; - if (alive && pid && run.providerProcessStartedAt) { + if (pidAlive && pid && run.providerProcessStartedAt) { const observedStartedAt = await readProcessStartedAt(pid).catch(() => null); ownsLiveProcess = observedStartedAt !== null && Date.parse(observedStartedAt) === run.providerProcessStartedAt.getTime(); + } else if (processGroupAlive && pid && processGroupId === pid) { + // The persisted guardian is the process-group leader. POSIX keeps that + // group id allocated while any original descendant remains. If the id + // has been reused by a new live leader, the branch above validates its + // start identity instead. + ownsLiveProcess = true; } if (ownsLiveProcess) { await terminateHeartbeatRunProcess({ pid, processGroupId }); @@ -16320,7 +16327,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) environment, onLog, onSpawn, - onProviderSpawn: async (meta) => { + onProviderOwnerSpawn: async (meta) => { await persistRunProviderProcessMetadata(run.id, meta); }, onProviderExit: async () => { diff --git a/server/src/services/native-runtime/native-codex-runner.ts b/server/src/services/native-runtime/native-codex-runner.ts index 6c54af54661..8e49cfea916 100644 --- a/server/src/services/native-runtime/native-codex-runner.ts +++ b/server/src/services/native-runtime/native-codex-runner.ts @@ -16,6 +16,15 @@ import { runnerPrpCoordinator } from "./runner-prp-coordinator.js"; const moduleDirectory = dirname(fileURLToPath(import.meta.url)); const RUNNER_VERSION = "paperclip-runner-v1"; const MAX_RUNNER_RESTARTS = 3; +const PROVIDER_GUARDIAN_SCRIPT = [ + "IFS= read -r ready || exit 0", + '[ "$ready" = "start" ] || exit 0', + '"$@" >/dev/null &', + "provider_pid=$!", + "printf '%s\\n' \"$provider_pid\"", + 'wait "$provider_pid" || true', + "while :; do sleep 3600; done", +].join("\n"); function executableName(): string { return process.platform === "win32" ? "paperclip-runnerd.exe" : "paperclip-runnerd"; @@ -126,6 +135,85 @@ async function waitForChildExit(exit: Promise, timeoutMs: number): Prom } } +async function startGuardedProvider(input: { + command: string; + args: string[]; + cwd: string; + environment: Record; + onOwnerSpawn?: (meta: { + pid: number; + processGroupId: number; + startedAt: string; + }) => Promise; +}): Promise<{ + guardian: ChildProcess; + exit: Promise<{ code: number | null; signal: NodeJS.Signals | null }>; + providerPid: number; +}> { + if (process.platform === "win32") { + throw new Error("codex_app_server_guardian_unsupported_platform"); + } + const guardian = spawn( + "/bin/sh", + ["-c", PROVIDER_GUARDIAN_SCRIPT, "paperclip-codex-provider", input.command, ...input.args], + { + cwd: input.cwd, + detached: true, + env: { ...process.env, ...input.environment }, + stdio: ["pipe", "pipe", "pipe"], + }, + ); + const exit = waitForExit(guardian); + try { + if (!guardian.pid) throw new Error("codex_app_server_guardian_not_started"); + const ownerStartedAt = await readProcessStartedAt(guardian.pid); + if (!ownerStartedAt) throw new Error("codex_app_server_guardian_identity_unavailable"); + await input.onOwnerSpawn?.({ + pid: guardian.pid, + processGroupId: guardian.pid, + startedAt: ownerStartedAt, + }); + + let buffer = ""; + const providerPid = await new Promise((resolvePid, rejectPid) => { + const timeout = setTimeout(() => { + cleanup(); + rejectPid(new Error("codex_app_server_guardian_start_timeout")); + }, 10_000); + timeout.unref(); + const cleanup = () => { + clearTimeout(timeout); + guardian.stdout?.off("data", onData); + }; + const onData = (chunk: Buffer) => { + buffer += chunk.toString("utf8"); + const newline = buffer.indexOf("\n"); + if (newline < 0) return; + const parsedPid = Number.parseInt(buffer.slice(0, newline).trim(), 10); + cleanup(); + if (!Number.isInteger(parsedPid) || parsedPid <= 0) { + rejectPid(new Error("codex_app_server_guardian_invalid_provider_pid")); + return; + } + resolvePid(parsedPid); + }; + guardian.stdout?.on("data", onData); + void exit.then(({ code, signal }) => { + cleanup(); + rejectPid( + new Error(`codex_app_server_guardian_exited: code=${code ?? "null"} signal=${signal ?? "null"}`), + ); + }, rejectPid); + guardian.stdin?.end("start\n"); + }); + return { guardian, exit, providerPid }; + } catch (error) { + guardian.stdin?.destroy(); + await stopChild(guardian, exit).catch(() => undefined); + throw error; + } +} + function signalRunnerProcessGroup(child: ChildProcess, signal: NodeJS.Signals): void { if (process.platform !== "win32" && child.pid) { try { @@ -197,6 +285,12 @@ export async function executeNativeCodexRunner(input: { processGroupId: number | null; startedAt: string; }) => Promise; + /** Persists the provider guardian before it is allowed to launch Codex. */ + onProviderOwnerSpawn?: (meta: { + pid: number; + processGroupId: number; + startedAt: string; + }) => Promise; /** Clears durable provider ownership after the shared server has stopped. */ onProviderExit?: () => Promise; }): Promise { @@ -269,31 +363,28 @@ export async function executeNativeCodexRunner(input: { if (useSharedCodexServer) { let startupError: Error | null = null; let stderr = ""; - sharedCodexServer = spawn( - "codex", - ["app-server", "--listen", `unix://${sharedCodexSocket}`], - { - cwd: input.cwd, - detached: process.platform !== "win32", - env: { ...process.env, ...input.environment }, - stdio: ["ignore", "ignore", "pipe"], - }, - ); - sharedCodexExit = waitForExit(sharedCodexServer); + const guardedProvider = await startGuardedProvider({ + command: "codex", + args: ["app-server", "--listen", `unix://${sharedCodexSocket}`], + cwd: input.cwd, + environment: input.environment, + onOwnerSpawn: input.onProviderOwnerSpawn, + }); + sharedCodexServer = guardedProvider.guardian; + sharedCodexExit = guardedProvider.exit; sharedCodexServer.once("error", (error) => { startupError = error; }); sharedCodexServer.stderr?.setEncoding("utf8").on("data", (chunk: string) => { stderr = `${stderr}${chunk}`.slice(-16_384); }); - if (!sharedCodexServer.pid) throw new Error("codex_app_server_process_not_started"); - const providerStartedAt = await readProcessStartedAt(sharedCodexServer.pid); + const providerStartedAt = await readProcessStartedAt(guardedProvider.providerPid); if (!providerStartedAt) { throw new Error("codex_app_server_process_identity_unavailable"); } await input.onProviderSpawn?.({ - pid: sharedCodexServer.pid, - processGroupId: process.platform === "win32" ? null : sharedCodexServer.pid, + pid: guardedProvider.providerPid, + processGroupId: sharedCodexServer.pid ?? null, startedAt: providerStartedAt, }); const startupDeadline = Date.now() + 10_000;