fix(runtime): persist native provider ownership

Record the shared Codex provider process on its heartbeat run so cancellation and orphan recovery can terminate it after a control-plane restart. Preserve live providers while their detached Runner owner remains healthy, and cover both recovery paths.

Co-authored-by: Codex <codex@openai.com>

packages/db/src/migrations/0229_wise_amphibian.sql

@@ -0,0 +1,3 @@
+ALTER TABLE "heartbeat_runs" ADD COLUMN "provider_process_pid" integer;--> statement-breakpoint
+ALTER TABLE "heartbeat_runs" ADD COLUMN "provider_process_group_id" integer;--> statement-breakpoint
+ALTER TABLE "heartbeat_runs" ADD COLUMN "provider_process_started_at" timestamp with time zone;

packages/db/src/migrations/meta/_journal.json

@@ -1590,6 +1590,13 @@
       "when": 1787773030696,
       "tag": "0228_polite_black_knight",
       "breakpoints": true
+    },
+    {
+      "idx": 229,
+      "version": "7",
+      "when": 1787775574417,
+      "tag": "0229_wise_amphibian",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/heartbeat_runs.ts

@@ -63,6 +63,9 @@ export const heartbeatRuns = pgTable(
     processPid: integer("process_pid"),
     processGroupId: integer("process_group_id"),
     processStartedAt: timestamp("process_started_at", { withTimezone: true }),
+    providerProcessPid: integer("provider_process_pid"),
+    providerProcessGroupId: integer("provider_process_group_id"),
+    providerProcessStartedAt: timestamp("provider_process_started_at", { withTimezone: true }),
     lastOutputAt: timestamp("last_output_at", { withTimezone: true }),
     lastOutputSeq: integer("last_output_seq").notNull().default(0),
     lastOutputStream: text("last_output_stream"),

server/src/__tests__/heartbeat-process-recovery.test.ts

@@ -488,6 +488,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     runStatus?: "running" | "queued" | "failed";
     processPid?: number | null;
     processGroupId?: number | null;
+    providerProcessPid?: number | null;
+    providerProcessGroupId?: number | null;
     processLossRetryCount?: number;
     includeIssue?: boolean;
     runErrorCode?: string | null;
@@ -548,6 +550,9 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
         : { ...(input?.contextSnapshot ?? {}), issueId },
       processPid: input?.processPid ?? null,
       processGroupId: input?.processGroupId ?? null,
+      providerProcessPid: input?.providerProcessPid ?? null,
+      providerProcessGroupId: input?.providerProcessGroupId ?? null,
+      providerProcessStartedAt: input?.providerProcessPid ? now : null,
       processLossRetryCount: input?.processLossRetryCount ?? 0,
       errorCode: input?.runErrorCode ?? null,
       error: input?.runError ?? null,
@@ -1204,13 +1209,17 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     expect(agent).toEqual({ status: "error", errorReason: "Adapter failed" });
   });
 
-  it("keeps a local run active when the recorded pid is still alive", async () => {
+  it("keeps a local run and its provider active when the recorded runner pid is still alive", async () => {
     const child = spawnAliveProcess();
+    const provider = spawnAliveProcess();
     childProcesses.add(child);
+    childProcesses.add(provider);
     expect(child.pid).toBeTypeOf("number");
+    expect(provider.pid).toBeTypeOf("number");
 
     const { runId, wakeupRequestId } = await seedRunFixture({
       processPid: child.pid ?? null,
+      providerProcessPid: provider.pid ?? null,
       includeIssue: false,
     });
     const heartbeat = heartbeatService(db);
@@ -1222,6 +1231,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     expect(run?.status).toBe("running");
     expect(run?.errorCode).toBe("process_detached");
     expect(run?.error).toContain(String(child.pid));
+    expect(run?.providerProcessPid).toBe(provider.pid);
+    expect(isPidAlive(provider.pid)).toBe(true);
 
     const wakeup = await db
       .select()
@@ -1231,6 +1242,31 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
     expect(wakeup?.status).toBe("claimed");
   });
 
+  it("terminates a detached provider when its owning runner is lost", async () => {
+    const provider = spawnAliveProcess();
+    childProcesses.add(provider);
+    expect(provider.pid).toBeTypeOf("number");
+
+    const { runId } = await seedRunFixture({
+      processPid: 999_999_999,
+      providerProcessPid: provider.pid ?? null,
+      includeIssue: false,
+    });
+    const heartbeat = heartbeatService(db);
+
+    const result = await heartbeat.reapOrphanedRuns();
+
+    expect(result.reaped).toBe(1);
+    expect(await waitForPidExit(provider.pid ?? 0)).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);

server/src/services/heartbeat.ts

@@ -9906,6 +9906,56 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
       .then((rows) => rows[0] ?? null);
   }
 
+  async function persistRunProviderProcessMetadata(
+    runId: string,
+    meta: { pid: number; processGroupId: number | null; startedAt: string },
+  ) {
+    const startedAt = new Date(meta.startedAt);
+    return db
+      .update(heartbeatRuns)
+      .set({
+        providerProcessPid: meta.pid,
+        providerProcessGroupId: meta.processGroupId,
+        providerProcessStartedAt: Number.isNaN(startedAt.getTime()) ? new Date() : startedAt,
+        updatedAt: new Date(),
+      })
+      .where(eq(heartbeatRuns.id, runId))
+      .returning()
+      .then((rows) => rows[0] ?? null);
+  }
+
+  async function clearRunProviderProcessMetadata(runId: string) {
+    return db
+      .update(heartbeatRuns)
+      .set({
+        providerProcessPid: null,
+        providerProcessGroupId: null,
+        providerProcessStartedAt: null,
+        updatedAt: new Date(),
+      })
+      .where(eq(heartbeatRuns.id, runId))
+      .returning()
+      .then((rows) => rows[0] ?? null);
+  }
+
+  async function terminateRunProviderProcess(
+    run: Pick<
+      typeof heartbeatRuns.$inferSelect,
+      "id" | "providerProcessPid" | "providerProcessGroupId"
+    >,
+  ): Promise<boolean> {
+    const pid = run.providerProcessPid;
+    const processGroupId = run.providerProcessGroupId;
+    const alive = isProcessAlive(pid) || isProcessGroupAlive(processGroupId);
+    if (alive) {
+      await terminateHeartbeatRunProcess({ pid, processGroupId });
+    }
+    if (pid || processGroupId) {
+      await clearRunProviderProcessMetadata(run.id);
+    }
+    return alive;
+  }
+
   async function clearDetachedRunWarning(runId: string) {
     const updated = await db
       .update(heartbeatRuns)
@@ -10812,6 +10862,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
     for (const { run, agent } of activeRuns) {
       const running = runningProcesses.get(run.id);
       try {
+        await terminateRunProviderProcess(run);
         if (running) {
           await terminateHeartbeatRunProcess({
             pid: running.child.pid ?? run.processPid,
@@ -13730,6 +13781,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
         continue;
       }
 
+      await terminateRunProviderProcess(run);
+
       let descendantOnlyCleanup = false;
       if (processGroupAlive) {
         descendantOnlyCleanup = true;
@@ -16260,6 +16313,12 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
             environment,
             onLog,
             onSpawn,
+            onProviderSpawn: async (meta) => {
+              await persistRunProviderProcessMetadata(run.id, meta);
+            },
+            onProviderExit: async () => {
+              await clearRunProviderProcessMetadata(run.id);
+            },
           });
         } else {
           const adapterContext = { ...context };
@@ -19268,6 +19327,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
 
     const running = runningProcesses.get(run.id);
     try {
+      await terminateRunProviderProcess(run);
       if (running) {
         await terminateHeartbeatRunProcess({
           pid: running.child.pid ?? run.processPid,
@@ -19342,18 +19402,22 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
       });
 
       const running = runningProcesses.get(run.id);
-      if (running) {
-        await terminateHeartbeatRunProcess({
-          pid: running.child.pid ?? run.processPid,
-          processGroupId: running.processGroupId ?? run.processGroupId,
-          graceMs: Math.max(1, running.graceSec) * 1000,
-        });
+      try {
+        await terminateRunProviderProcess(run);
+        if (running) {
+          await terminateHeartbeatRunProcess({
+            pid: running.child.pid ?? run.processPid,
+            processGroupId: running.processGroupId ?? run.processGroupId,
+            graceMs: Math.max(1, running.graceSec) * 1000,
+          });
+        } else if (run.processPid || run.processGroupId) {
+          await terminateHeartbeatRunProcess({
+            pid: run.processPid,
+            processGroupId: run.processGroupId,
+          });
+        }
+      } finally {
         runningProcesses.delete(run.id);
-      } else if (run.processPid || run.processGroupId) {
-        await terminateHeartbeatRunProcess({
-          pid: run.processPid,
-          processGroupId: run.processGroupId,
-        });
       }
       await releaseIssueExecutionAndPromote(run);
     }

server/src/services/native-runtime/native-codex-runner.ts

@@ -191,7 +191,13 @@ export async function executeNativeCodexRunner(input: {
     startedAt: string;
   }) => Promise<void>;
   /** Internal qualification seam for recording the shared Codex server identity. */
-  onProviderSpawn?: (meta: { pid: number; startedAt: string }) => Promise<void>;
+  onProviderSpawn?: (meta: {
+    pid: number;
+    processGroupId: number | null;
+    startedAt: string;
+  }) => Promise<void>;
+  /** Clears durable provider ownership after the shared server has stopped. */
+  onProviderExit?: () => Promise<void>;
 }): Promise<AdapterExecutionResult> {
   const binary = input.runnerBinary ?? resolvePaperclipRunnerBinary();
   const runnerDigest = `sha256:${createHash("sha256").update(readFileSync(binary)).digest("hex")}`;
@@ -282,6 +288,7 @@ export async function executeNativeCodexRunner(input: {
       if (!sharedCodexServer.pid) throw new Error("codex_app_server_process_not_started");
       await input.onProviderSpawn?.({
         pid: sharedCodexServer.pid,
+        processGroupId: process.platform === "win32" ? null : sharedCodexServer.pid,
         startedAt: new Date().toISOString(),
       });
       const startupDeadline = Date.now() + 10_000;
@@ -409,6 +416,7 @@ export async function executeNativeCodexRunner(input: {
     if (sharedCodexServer && sharedCodexExit) {
       await stopChild(sharedCodexServer, sharedCodexExit).catch(() => undefined);
     }
+    await input.onProviderExit?.().catch(() => undefined);
     await prepared.release();
     if (sharedCodexSocketDirectory) {
       await rm(sharedCodexSocketDirectory, { recursive: true, force: true });