From cceda34670e91629536aafb306fed7d75a3d4251 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 18 Aug 2026 18:22:44 +0000 Subject: [PATCH] fix(runner): persist native process ownership Co-Authored-By: Paperclip --- .../src/backends/codex-native-backend.ts | 10 +++- .../codex/app-server-transport.test.ts | 11 +++++ .../src/drivers/codex/app-server-transport.ts | 4 ++ .../codex/codex-app-server-driver.test.ts | 39 +++++++++++++++- .../drivers/codex/codex-app-server-driver.ts | 20 ++++++++ .../src/live/clean-room-server.test.ts | 2 +- .../src/live/live-session.test.ts | 1 + .../src/live/turn-stream.test.ts | 2 +- .../heartbeat-process-recovery.test.ts | 20 ++++++++ server/src/services/heartbeat.ts | 35 +++++++++----- .../native-session-executor.test.ts | 46 ++++++++++++++++++- .../native-runtime/native-session-executor.ts | 10 +++- 12 files changed, 182 insertions(+), 18 deletions(-) diff --git a/packages/paperclip-runner/src/backends/codex-native-backend.ts b/packages/paperclip-runner/src/backends/codex-native-backend.ts index ee6df650ef4..68d39d3fc16 100644 --- a/packages/paperclip-runner/src/backends/codex-native-backend.ts +++ b/packages/paperclip-runner/src/backends/codex-native-backend.ts @@ -10,7 +10,14 @@ import { HarnessDriverBackend } from "./harness-driver-backend.js"; */ export function createCodexNativeSessionBackend( input: NativeExecutionInputV1, - options: { runnerInstanceId?: string } = {}, + options: { + runnerInstanceId?: string; + onSpawn?: (meta: { + pid: number; + processGroupId: number | null; + startedAt: string; + }) => Promise; + } = {}, ): NativeSessionBackend { return new HarnessDriverBackend(new CodexAppServerDriver({ taskEnvelope: createCodexTaskEnvelope({ @@ -25,5 +32,6 @@ export function createCodexNativeSessionBackend( ], }), runnerInstanceId: options.runnerInstanceId ?? `paperclip-native-${input.binding.runId}`, + onSpawn: options.onSpawn, })); } diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts index ed424c0daa9..f340b320d4a 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts @@ -15,6 +15,17 @@ function nodeTransport( } describe("Codex app-server transport limits", () => { + it("reports restart-safe process-group ownership", async () => { + const transport = nodeTransport("process.stdin.resume()", { processGroup: true }); + const info = transport.processInfo(); + + expect(info.pid).toBeGreaterThan(0); + expect(info.processGroupId).toBe(process.platform === "win32" ? null : info.pid); + expect(new Date(info.startedAt).toISOString()).toBe(info.startedAt); + + await transport.close(); + }); + it("rejects an oversized line before buffering the complete hostile payload", async () => { const diagnostics: string[] = []; const transport = nodeTransport( diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index 541d4905ec8..6bb6fcd21bb 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -25,6 +25,7 @@ export interface CodexAppServerTransport { export interface CodexTransportProcessInfo { pid: number | null; processGroupId: number | null; + startedAt: string; exited: boolean; exitCode: number | null; signal: NodeJS.Signals | null; @@ -300,6 +301,7 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { #resolveExit!: () => void; readonly #exitPromise: Promise; readonly #processGroup: boolean; + readonly #startedAt: string; readonly #closeGraceMs: number; readonly #onProcess?: (info: CodexTransportProcessInfo) => void; #onDiagnostic?: (message: string) => void; @@ -321,6 +323,7 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS, ); this.#processGroup = options.processGroup === true && process.platform !== "win32"; + this.#startedAt = new Date().toISOString(); this.#closeGraceMs = positiveLimit(options.closeGraceMs, 1_000); this.#onProcess = options.onProcess; this.#exitPromise = new Promise((resolve) => { @@ -430,6 +433,7 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { return { pid: this.#process.pid ?? null, processGroupId: this.#processGroup ? this.#process.pid ?? null : null, + startedAt: this.#startedAt, exited: this.#exited, exitCode: this.#exitCode, signal: this.#exitSignal, diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts index fd6431a06bf..a67baab5025 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { fileURLToPath } from "node:url"; import { @@ -272,6 +272,43 @@ async function traceCompletedProposal( } describe("Codex app-server Codex driver", () => { + it("persists process ownership before sending provider requests", async () => { + const transport = new FakeCodexTransport(); + Object.assign(transport, { + processInfo: () => ({ + pid: 71_001, + processGroupId: 71_001, + startedAt: "2026-08-18T18:00:00.000Z", + exited: false, + exitCode: null, + signal: null, + }), + }); + let releaseOwnership!: () => void; + const ownershipPersisted = new Promise((resolve) => { + releaseOwnership = resolve; + }); + const onSpawn = vi.fn(() => ownershipPersisted); + const driver = makeDriver([transport], { onSpawn }); + + const opening = driver.openSession({ + runId: "run-owned", + normalizedSessionId: "normalized-owned", + workingDirectory: "/workspace", + }); + await vi.waitFor(() => expect(onSpawn).toHaveBeenCalledOnce()); + expect(transport.calls).toEqual([]); + + releaseOwnership(); + await opening; + expect(onSpawn).toHaveBeenCalledWith({ + pid: 71_001, + processGroupId: 71_001, + startedAt: "2026-08-18T18:00:00.000Z", + }); + expect(transport.calls[0]?.method).toBe("initialize"); + }); + it("sends direct chat as plain text and permits a follow-up turn", async () => { const transport = new FakeCodexTransport(); const driver = makeDriver([transport], { conversationMode: "direct" }); diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts index c8ce44a4cb2..31c6745fe32 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.ts @@ -49,6 +49,7 @@ import { type CodexAppServerTransport, type CodexRpcNotification, type CodexRpcServerRequest, + type CodexTransportProcessInfo, } from "./app-server-transport.js"; const DRIVER_KIND = "codex_app_server"; @@ -95,6 +96,11 @@ export interface CodexAppServerDriverOptions { now?: () => Date; runnerInstanceId?: string; onDiagnostic?: (message: string) => void; + onSpawn?: (meta: { + pid: number; + processGroupId: number | null; + startedAt: string; + }) => Promise; capabilities?: Partial<{ resume: boolean; read: boolean; @@ -345,6 +351,7 @@ export class CodexAppServerDriver implements HarnessDriver { const workingDirectory = validateWorkingDirectory(input.workingDirectory, this.#options.environment); const transport = this.#transport(); try { + await this.#persistProcessOwnership(transport); const initialize = await this.#initialize(transport); const response = await transport.request("thread/start", { ...securedThreadParams(workingDirectory, this.#options.environment), @@ -388,6 +395,7 @@ export class CodexAppServerDriver implements HarnessDriver { } const transport = this.#transport(); try { + await this.#persistProcessOwnership(transport); const initialize = await this.#initialize(transport); const existing = await transport.request("thread/read", { threadId: snapshot.driverSessionId, @@ -452,10 +460,22 @@ export class CodexAppServerDriver implements HarnessDriver { args: createIsolatedCodexAppServerArgs(this.#options.environment), environment: createSanitizedCodexEnvironment(this.#options.environment), onDiagnostic: this.#options.onDiagnostic, + processGroup: true, }) ); } + async #persistProcessOwnership(transport: CodexAppServerTransport): Promise { + if (!this.#options.onSpawn) return; + const processInfo: CodexTransportProcessInfo | undefined = transport.processInfo?.(); + if (!processInfo || processInfo.exited || processInfo.pid === null) return; + await this.#options.onSpawn({ + pid: processInfo.pid, + processGroupId: processInfo.processGroupId, + startedAt: processInfo.startedAt, + }); + } + async #initialize(transport: CodexAppServerTransport): Promise> { const initialized = await transport.request("initialize", { clientInfo: { name: "paperclip-runner", title: "Paperclip Runner", version: DRIVER_VERSION }, diff --git a/packages/paperclip-runner/src/live/clean-room-server.test.ts b/packages/paperclip-runner/src/live/clean-room-server.test.ts index 982ee748e16..d491d801d22 100644 --- a/packages/paperclip-runner/src/live/clean-room-server.test.ts +++ b/packages/paperclip-runner/src/live/clean-room-server.test.ts @@ -168,7 +168,7 @@ class FakeCleanRoomTransport implements CodexAppServerTransport { } processInfo() { - return { pid: 7100, processGroupId: 7100, exited: this.#closed, exitCode: null, signal: null }; + return { pid: 7100, processGroupId: 7100, startedAt: "2026-08-01T00:00:00.000Z", exited: this.#closed, exitCode: null, signal: null }; } async #runTurn(turnId: string, message: string): Promise { diff --git a/packages/paperclip-runner/src/live/live-session.test.ts b/packages/paperclip-runner/src/live/live-session.test.ts index db298e146ef..185900756e6 100644 --- a/packages/paperclip-runner/src/live/live-session.test.ts +++ b/packages/paperclip-runner/src/live/live-session.test.ts @@ -123,6 +123,7 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { return { pid: 4100 + this.state.transports.indexOf(this), processGroupId: 4100 + this.state.transports.indexOf(this), + startedAt: "2026-08-01T00:00:00.000Z", exited: this.#closed, exitCode: this.#closed ? 0 : null, signal: null, diff --git a/packages/paperclip-runner/src/live/turn-stream.test.ts b/packages/paperclip-runner/src/live/turn-stream.test.ts index 68b0906f0aa..05ec8e8678b 100644 --- a/packages/paperclip-runner/src/live/turn-stream.test.ts +++ b/packages/paperclip-runner/src/live/turn-stream.test.ts @@ -126,7 +126,7 @@ class GatedTransport implements CodexAppServerTransport { } processInfo() { - return { pid: 7100, processGroupId: 7100, exited: this.#closed, exitCode: null, signal: null }; + return { pid: 7100, processGroupId: 7100, startedAt: "2026-08-01T00:00:00.000Z", exited: this.#closed, exitCode: null, signal: null }; } async #runTurn(turnId: string): Promise { diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index ad0ed77cb29..57f6a0c8de6 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -103,6 +103,7 @@ import { INTERACTION_CONTINUATION_INFRA_RETRY_REASON, INTERACTION_CONTINUATION_INFRA_WAKE_REASON, heartbeatService, + persistHeartbeatRunProcessMetadata, redactDetectedSuccessfulRunProgressSummaryForBoard, redactSuccessfulRunHandoffEvidence, } from "../services/heartbeat.ts"; @@ -573,6 +574,25 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { return { companyId, agentId, runId, wakeupRequestId, issueId }; } + it("persists native app-server ownership for restart-time cancellation", async () => { + const { runId } = await seedRunFixture({ includeIssue: false }); + const startedAt = "2026-08-18T18:00:00.000Z"; + + await persistHeartbeatRunProcessMetadata(db, runId, { + pid: 81_001, + processGroupId: 81_001, + startedAt, + }); + + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId))).resolves.toEqual([ + expect.objectContaining({ + processPid: 81_001, + processGroupId: 81_001, + processStartedAt: new Date(startedAt), + }), + ]); + }); + async function seedEnvironmentLeaseFixture(input: { companyId: string; runId: string; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index ca9119af598..768bcf2f699 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -6318,6 +6318,25 @@ function isProcessAlive(pid: number | null | undefined) { } } +export async function persistHeartbeatRunProcessMetadata( + db: Db, + runId: string, + meta: { pid: number; processGroupId: number | null; startedAt: string }, +) { + const startedAt = new Date(meta.startedAt); + return db + .update(heartbeatRuns) + .set({ + processPid: meta.pid, + processGroupId: meta.processGroupId, + processStartedAt: Number.isNaN(startedAt.getTime()) ? new Date() : startedAt, + updatedAt: new Date(), + }) + .where(eq(heartbeatRuns.id, runId)) + .returning() + .then((rows) => rows[0] ?? null); +} + async function terminateHeartbeatRunProcess(input: { pid: number | null | undefined; processGroupId: number | null | undefined; @@ -9813,18 +9832,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) runId: string, meta: { pid: number; processGroupId: number | null; startedAt: string }, ) { - const startedAt = new Date(meta.startedAt); - return db - .update(heartbeatRuns) - .set({ - processPid: meta.pid, - processGroupId: meta.processGroupId, - processStartedAt: Number.isNaN(startedAt.getTime()) ? new Date() : startedAt, - updatedAt: new Date(), - }) - .where(eq(heartbeatRuns.id, runId)) - .returning() - .then((rows) => rows[0] ?? null); + return persistHeartbeatRunProcessMetadata(db, runId, meta); } async function clearDetachedRunWarning(runId: string) { @@ -16150,6 +16158,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) runnerInstanceId: nativeRunnerInstanceId, leaseOwner: runOptions.nativeLeaseOwner, backend: options.nativeSessionBackendFactory?.(nativeExecution), + onSpawn: async (meta) => { + await persistRunProcessMetadata(run.id, meta); + }, }); } else { const adapterContext = { ...context }; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 7556ac08abe..2a020c9c05b 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -4,12 +4,13 @@ import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner"; const state = vi.hoisted(() => ({ execute: vi.fn(), + createBackend: vi.fn(() => ({ kind: "test" })), cancel: vi.fn(), release: null as null | (() => void), })); vi.mock("@paperclipai/paperclip-runner", () => ({ - createCodexNativeSessionBackend: vi.fn(() => ({ kind: "test" })), + createCodexNativeSessionBackend: state.createBackend, executeNativeSession: state.execute, })); @@ -141,3 +142,46 @@ describe("native session bounded recovery", () => { }); }); }); + +describe("native process ownership", () => { + it("forwards the app-server PID and process group through the production backend seam", async () => { + const processMetadata = { + pid: 42_001, + processGroupId: 42_001, + startedAt: "2026-08-18T18:00:00.000Z", + }; + const onSpawn = vi.fn(async () => undefined); + state.createBackend.mockClear(); + state.execute.mockReset().mockImplementation(async (options) => { + await options.backend.onSpawn(processMetadata); + return { + result: { summary: "completed" }, + terminal: { runTerminalState: "succeeded" }, + turnId: "turn", + normalizedSessionId: "session", + providerSessionId: null, + driverKind: "test", + driverVersion: "1", + nativeEventCount: 1, + highestContiguousSourceSeq: 1, + }; + }); + state.createBackend.mockImplementationOnce((_input, options) => ({ + kind: "test", + onSpawn: options.onSpawn, + })); + + await executePaperclipNativeSession({ + db: leaseDb(), + execution, + runnerInstanceId: "runner", + onSpawn, + }); + + expect(state.createBackend).toHaveBeenCalledWith(execution, { + runnerInstanceId: "runner", + onSpawn, + }); + expect(onSpawn).toHaveBeenCalledWith(processMetadata); + }); +}); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 5a87d0cb9e4..902ec335fad 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -234,6 +234,11 @@ export async function executePaperclipNativeSession(input: { execution: NativeExecutionInputV1; runnerInstanceId: string; leaseOwner?: string; + onSpawn?: (meta: { + pid: number; + processGroupId: number | null; + startedAt: string; + }) => Promise; /** Test seam at the provider boundary; production always uses the package Codex backend. */ backend?: NativeSessionBackend; }): Promise { @@ -288,7 +293,10 @@ export async function executePaperclipNativeSession(input: { native = await executeNativeSession({ input: input.execution, backend: input.backend - ?? createCodexNativeSessionBackend(input.execution, { runnerInstanceId: input.runnerInstanceId }), + ?? createCodexNativeSessionBackend(input.execution, { + runnerInstanceId: input.runnerInstanceId, + onSpawn: input.onSpawn, + }), controlPlane, runnerInstanceId: input.runnerInstanceId, controlPlaneInstanceId,