From a87c6b41cd0bcfe23e0c16e0a49a2efe7c2a70af Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 18 Jul 2026 19:56:42 +0000 Subject: [PATCH] fix(heartbeat): persist codex process identity before execution Co-Authored-By: Paperclip --- .../src/acpx-engine/execute.test.ts | 68 ++++++++++++++ .../adapter-utils/src/acpx-engine/execute.ts | 25 ++++- .../adapter-utils/src/server-utils.test.ts | 25 +++++ packages/adapter-utils/src/server-utils.ts | 23 +++-- .../heartbeat-process-recovery.test.ts | 91 +++++++++++++++++++ 5 files changed, 223 insertions(+), 9 deletions(-) diff --git a/packages/adapter-utils/src/acpx-engine/execute.test.ts b/packages/adapter-utils/src/acpx-engine/execute.test.ts index 56e41dc6feb..b16a2b926b8 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.test.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.test.ts @@ -161,6 +161,74 @@ async function runExecutor( } describe("shared ACPX engine runtime behavior", () => { + it("reports the local ACP child before starting the first Codex turn", async () => { + const root = await makeTempRoot(); + const stateDir = path.join(root, "state"); + const processStartedAt = "2026-07-18T12:00:00.000Z"; + const spawnEvents: Array<{ + pid: number; + processGroupId: number | null; + startedAt: string; + }> = []; + let turnStartedAfterSpawn = false; + const execute = createAcpxEngineExecutor({ + adapterType: "codex_local", + createRuntime: (options) => { + const loadSession = options.sessionStore.load.bind(options.sessionStore); + options.sessionStore.load = async (recordId) => + recordId === "record-1" + ? ({ pid: process.pid, agentStartedAt: processStartedAt } as never) + : loadSession(recordId); + return { + ensureSession: async () => ({ + sessionKey: "session-1", + acpxRecordId: "record-1", + backendSessionId: "backend-session", + agentSessionId: "agent-session", + runtimeSessionName: "runtime-session", + }), + startTurn: () => { + turnStartedAfterSpawn = spawnEvents.length === 1; + return { + events: (async function* () { + yield { type: "done", stopReason: "end_turn" }; + })(), + result: Promise.resolve({ status: "completed", stopReason: "end_turn" }), + cancel: async () => {}, + }; + }, + close: async () => {}, + } as never; + }, + }); + + const result = await execute({ + runId: "run-codex-process-metadata", + agent: { id: "agent-1", companyId: "company-1" }, + runtime: {}, + config: { agent: "codex", stateDir, cwd: root, mode: "oneshot" }, + context: {}, + onLog: async () => {}, + onSpawn: async (meta: { + pid: number; + processGroupId: number | null; + startedAt: string; + }) => { + spawnEvents.push(meta); + }, + } as never); + + expect(result.exitCode).toBe(0); + expect(spawnEvents).toEqual([ + { + pid: process.pid, + processGroupId: null, + startedAt: processStartedAt, + }, + ]); + expect(turnStartedAfterSpawn).toBe(true); + }); + it("sets Codex model, effort, and fast mode through CODEX_CONFIG without session config calls", async () => { const { configOptions, meta } = await runExecutor({ agent: "codex", diff --git a/packages/adapter-utils/src/acpx-engine/execute.ts b/packages/adapter-utils/src/acpx-engine/execute.ts index d23841e35f9..6171db64e89 100644 --- a/packages/adapter-utils/src/acpx-engine/execute.ts +++ b/packages/adapter-utils/src/acpx-engine/execute.ts @@ -60,6 +60,7 @@ import { type AcpRuntimeEvent, type AcpRuntimeHandle, type AcpRuntimeOptions, + type AcpSessionStore, type AcpRuntimeStatus, type AcpRuntimeTurn, type AcpRuntimeTurnResult, @@ -1927,6 +1928,21 @@ function warmHandleMatches( return entry !== undefined && entry.runtime === runtime && entry.handle === handle; } +async function persistLocalAcpProcessMetadata(input: { + ctx: AdapterExecutionContext; + prepared: AcpxPreparedRuntime; + sessionStore: AcpSessionStore; + handle: AcpRuntimeHandle; +}) { + if (!input.ctx.onSpawn || input.prepared.remoteExecutionIdentity) return; + const recordId = input.handle.acpxRecordId ?? input.handle.sessionKey; + const record = await input.sessionStore.load(recordId); + const pid = record?.pid; + if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 0) return; + const startedAt = record?.agentStartedAt ?? record?.createdAt ?? new Date().toISOString(); + await input.ctx.onSpawn({ pid, processGroupId: null, startedAt }); +} + export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const createRuntime = deps.createRuntime ?? createAcpRuntime; const now = deps.now ?? (() => Date.now()); @@ -1961,9 +1977,10 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { const canResume = isCompatibleSession(previousParams, prepared); const resumeSessionId = canResume ? asString(previousParams.acpSessionId, "") || undefined : undefined; const cached = canResume ? warmHandles.get(prepared.sessionKey) : undefined; + const sessionStore = createRuntimeStore({ stateDir: prepared.stateDir }); const runtimeOptions: AcpRuntimeOptions = { cwd: prepared.cwd, - sessionStore: createRuntimeStore({ stateDir: prepared.stateDir }), + sessionStore, agentRegistry: prepared.agentRegistry, permissionMode: prepared.permissionMode, nonInteractivePermissions: prepared.nonInteractivePermissions, @@ -2051,6 +2068,12 @@ export function createAcpxEngineExecutor(deps: AcpxEngineExecutorOptions = {}) { } const sessionHandle = handle; try { + await persistLocalAcpProcessMetadata({ + ctx, + prepared, + sessionStore, + handle: sessionHandle, + }); await applySessionConfigOptions({ runtime, handle: sessionHandle, diff --git a/packages/adapter-utils/src/server-utils.test.ts b/packages/adapter-utils/src/server-utils.test.ts index 4ce82bbbe6d..131b3460210 100644 --- a/packages/adapter-utils/src/server-utils.test.ts +++ b/packages/adapter-utils/src/server-utils.test.ts @@ -442,6 +442,31 @@ describe("runChildProcess", () => { expect(finishedAt - startedAt).toBeGreaterThanOrEqual(spawnDelayMs); }); + it("does not send stdin when onSpawn persistence fails", async () => { + const run = runChildProcess( + randomUUID(), + process.execPath, + [ + "-e", + "process.stdin.on('data',()=>process.stdout.write('unexpected'));setInterval(()=>{},1000);", + ], + { + cwd: process.cwd(), + env: {}, + stdin: "restart server", + timeoutSec: 5, + graceSec: 1, + onLog: async () => {}, + onLogError: () => {}, + onSpawn: async () => { + throw new Error("database unavailable"); + }, + }, + ); + + await expect(run).rejects.toThrow("database unavailable"); + }); + it.skipIf(process.platform === "win32")("kills descendant processes on timeout via the process group", async () => { let descendantPid: number | null = null; diff --git a/packages/adapter-utils/src/server-utils.ts b/packages/adapter-utils/src/server-utils.ts index d3e6b4e73ad..e17491e1588 100644 --- a/packages/adapter-utils/src/server-utils.ts +++ b/packages/adapter-utils/src/server-utils.ts @@ -3069,9 +3069,7 @@ export async function runChildProcess( const spawnPersistPromise = typeof child.pid === "number" && child.pid > 0 && opts.onSpawn - ? opts.onSpawn({ pid: child.pid, processGroupId, startedAt }).catch((err) => { - onLogError(err, runId, "failed to record child process metadata"); - }) + ? opts.onSpawn({ pid: child.pid, processGroupId, startedAt }) : Promise.resolve(); runningProcesses.set(runId, { child, graceSec: opts.graceSec, processGroupId }); @@ -3179,12 +3177,21 @@ export async function runChildProcess( }); const stdin = child.stdin; + const handleSpawnPersistenceFailure = (err: unknown) => { + onLogError(err, runId, "failed to record child process metadata"); + signalRunningProcess({ child, processGroupId }, "SIGKILL"); + reject(err instanceof Error ? err : new Error(String(err))); + }; if (opts.stdin != null && stdin) { - void spawnPersistPromise.finally(() => { - if (child.killed || stdin.destroyed) return; - stdin.write(opts.stdin as string); - stdin.end(); - }); + void spawnPersistPromise + .then(() => { + if (child.killed || stdin.destroyed) return; + stdin.write(opts.stdin as string); + stdin.end(); + }) + .catch(handleSpawnPersistenceFailure); + } else { + void spawnPersistPromise.catch(handleSpawnPersistenceFailure); } child.on("error", (err: Error) => { diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 1432021f91e..4a9da582ac0 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1555,6 +1555,97 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("persists codex_local onSpawn metadata before hot-restart adoption", async () => { + const child = spawnAliveProcess(); + childProcesses.add(child); + expect(child.pid).toBeGreaterThan(0); + const processStartedAt = "2026-03-19T00:05:30.000Z"; + let releaseAdapter: (() => void) | null = null; + let markAdapterSpawned: (() => void) | null = null; + const adapterSpawned = new Promise((resolve) => { + markAdapterSpawned = resolve; + }); + const adapterRelease = new Promise((resolve) => { + releaseAdapter = resolve; + }); + mockAdapterExecute.mockImplementationOnce(async (ctx: { + onSpawn?: (meta: { + pid: number; + processGroupId: number | null; + startedAt: string; + }) => Promise; + }) => { + await ctx.onSpawn?.({ + pid: child.pid!, + processGroupId: null, + startedAt: processStartedAt, + }); + markAdapterSpawned?.(); + await adapterRelease; + return { + exitCode: 0, + signal: null, + timedOut: false, + errorMessage: null, + summary: "Codex run completed after hot-restart adoption.", + provider: "test", + model: "test-model", + }; + }); + + const { runId } = await seedRunFixture({ + adapterType: "codex_local", + agentStatus: "idle", + runStatus: "queued", + processPid: null, + processGroupId: null, + includeIssue: false, + }); + const heartbeat = heartbeatService(db); + + try { + await heartbeat.resumeQueuedRuns(); + await adapterSpawned; + + const running = await heartbeat.getRun(runId); + expect(running).toMatchObject({ + status: "running", + processPid: child.pid, + processGroupId: null, + }); + expect(running?.processStartedAt?.toISOString()).toBe(processStartedAt); + + await withTempPaperclipHome(async () => { + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-version", + requestedAt: new Date("2026-03-19T00:05:45.000Z"), + }); + await heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-03-19T00:06:00.000Z"), + ); + + const adoption = await heartbeat.reconcileHotRestartAdoption( + new Date("2026-03-19T00:07:00.000Z"), + ); + expect(adoption).toMatchObject({ + mode: "reported", + adoptedRunIds: [runId], + finalizedWhileDownRunIds: [], + lostRunIds: [], + skippedRunIds: [], + }); + }); + + const adopted = await heartbeat.getRun(runId); + expect(adopted?.errorCode).not.toBe("process_lost"); + } finally { + releaseAdapter?.(); + await waitForRunToSettle(heartbeat, runId, 5_000); + } + }); + it.skipIf(process.platform === "win32")("keeps process-group-only hot-restart adoptions out of process_lost reaping", async () => { const orphan = await spawnOrphanedProcessGroup(); cleanupPids.add(orphan.descendantPid);