fix(heartbeat): persist codex process identity before execution

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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",

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,

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;
 

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) => {

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<void>((resolve) => {
+      markAdapterSpawned = resolve;
+    });
+    const adapterRelease = new Promise<void>((resolve) => {
+      releaseAdapter = resolve;
+    });
+    mockAdapterExecute.mockImplementationOnce(async (ctx: {
+      onSpawn?: (meta: {
+        pid: number;
+        processGroupId: number | null;
+        startedAt: string;
+      }) => Promise<void>;
+    }) => {
+      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);