fix(runner): persist native process ownership

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-18 18:22:44 +00:00
1 parent 78291feeb5
commit cceda34670
12 files changed
+182 -18

No files matched your search

@@ -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<void>;
} = {},
): 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,
}));
}
@@ -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(
@@ -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<void>;
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<void>((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,
@@ -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<void>((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" });
@@ -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<void>;
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<void> {
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<Record<string, unknown>> {
const initialized = await transport.request("initialize", {
clientInfo: { name: "paperclip-runner", title: "Paperclip Runner", version: DRIVER_VERSION },
@@ -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<void> {
@@ -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,
@@ -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<void> {
@@ -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;
+23 -12
View File
@@ -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 };
@@ -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);
});
});
@@ -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<void>;
/** Test seam at the provider boundary; production always uses the package Codex backend. */
backend?: NativeSessionBackend;
}): Promise<AdapterExecutionResult> {
@@ -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,