From 6010158ecc5affe881bd74cff5636643641a8429 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 23 Aug 2026 20:15:48 +0000 Subject: [PATCH] fix(runner): rejoin a shared Codex app-server Co-Authored-By: Paperclip --- .../durable-prp-control-plane.ts | 94 ++++- .../paperclip-runner/src/live/live-session.ts | 1 + .../src/mock-core/durable-recovery.test.ts | 15 +- .../fake-native-resume-codex-app-server.mjs | 339 ++++++++++-------- 4 files changed, 289 insertions(+), 160 deletions(-) diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index ab39539b90f..9d224716f39 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -11,6 +11,7 @@ import { chmodSync, closeSync, constants, + existsSync, fchmodSync, fstatSync, fsyncSync, @@ -1992,6 +1993,10 @@ export interface DurableEvalSessionInput { /** Test-only provider process override; production eval callers omit these. */ providerCommand?: string; providerArgs?: string[]; + /** Test-only shared Codex server override; production eval callers omit these. */ + sharedProviderCommand?: string; + sharedProviderArgs?: string[]; + sharedProviderSocketPath?: string; /** Defaults on for Codex; set false only for an eval that needs the baseline prompt. */ includeCollaborationModeInstructions?: boolean; provider?: "codex" | "opencode" | "claude_managed" | "aws_agentcore" | "acpx"; @@ -2417,6 +2422,23 @@ export async function runDurableEvalSession( providerKind === "acpx" ? resolveQualifiedAcpxProfile(input.acpxAgent ?? "pi", input.model) : null; + const sharedCodexSocket = + input.sharedProviderSocketPath ?? resolve(root, "c.sock"); + const codexConfigArgs = providerArgs.slice(0, -1); + const effectiveProviderArgs = + input.providerArgs ?? + (input.nativeResume === undefined + ? providerArgs + : [ + ...codexConfigArgs, + "app-server", + "proxy", + "--sock", + sharedCodexSocket, + ]); + let sharedCodexServer: ChildProcessWithoutNullStreams | null = null; + let sharedCodexServerStderr = ""; + let sharedCodexServerError: Error | null = null; core.queueCommand("run.prepare", { authorizedTools: { schema: "paperclip.runner.authorized-tools.v1", @@ -2468,7 +2490,7 @@ export async function runDurableEvalSession( input.providerArgs ?? (providerKind === "opencode" ? [opencodeProxyPath] - : providerArgs), + : effectiveProviderArgs), cwd: tmpdir(), model: input.model, instructions: @@ -2482,6 +2504,49 @@ export async function runDurableEvalSession( core.queueCommand("session.open", { reuse: "same_session" }); core.queueCommand("turn.start", { text: input.prompt }); await core.start(); + if (input.nativeResume !== undefined) { + const sharedCommand = + input.sharedProviderCommand ?? + input.providerCommand ?? + process.env.PAPERCLIP_CODEX_COMMAND ?? + "codex"; + const sharedArgs = + input.sharedProviderArgs ?? + [ + ...codexConfigArgs, + "app-server", + "--listen", + `unix://${sharedCodexSocket}`, + ]; + sharedCodexServer = spawn(sharedCommand, sharedArgs, { + cwd: tmpdir(), + env: process.env, + stdio: "pipe", + }); + sharedCodexServer.stdout.resume(); + sharedCodexServer.once("error", (error) => { + sharedCodexServerError = error; + }); + sharedCodexServer.stderr.setEncoding("utf8").on("data", (chunk: string) => { + sharedCodexServerStderr = `${sharedCodexServerStderr}${chunk}`.slice(-16_384); + }); + const sharedServerDeadline = Date.now() + 10_000; + while ( + !existsSync(sharedCodexSocket) && + sharedCodexServerError === null && + sharedCodexServer.exitCode === null && + Date.now() < sharedServerDeadline + ) { + await new Promise((resolveWait) => setTimeout(resolveWait, 20)); + } + if (!existsSync(sharedCodexSocket)) { + sharedCodexServer.kill("SIGKILL"); + const startError = sharedCodexServerError as Error | null; + throw new Error( + `shared Codex app-server did not create its socket: ${startError?.message ?? sharedCodexServerStderr}`, + ); + } + } const launchRunner = (): RunnerProcessHandle => spawnRunner({ connectUrl: core.connectUrl, @@ -2969,12 +3034,13 @@ export async function runDurableEvalSession( comments?: unknown[]; }; const finalComments = authority.snapshot().comments; - return { - schema: "paperclip.runner.native-resume-proof/v1", - triggered: true, - runnerRestarts, - runnerProcessPids, - providerProcessPids, + return { + schema: "paperclip.runner.native-resume-proof/v1", + triggered: true, + runnerRestarts, + runnerProcessPids, + providerProcessPids, + sharedCodexServerPid: sharedCodexServer?.pid ?? null, sameProviderThread: providerThreadIds.length >= 2 && new Set(providerThreadIds).size === 1, @@ -3166,6 +3232,20 @@ export async function runDurableEvalSession( } finally { releaseNativeSemanticDispatch(); handle.child.kill("SIGKILL"); + const sharedServer = sharedCodexServer; + if (sharedServer !== null && sharedServer.exitCode === null) { + sharedServer.kill("SIGTERM"); + await new Promise((resolveExit) => { + const timer = setTimeout(() => { + sharedServer.kill("SIGKILL"); + resolveExit(); + }, 2_000); + sharedServer.once("exit", () => { + clearTimeout(timer); + resolveExit(); + }); + }); + } await core.stop(); await authority.stop(); if (process.env.PAPERCLIP_KEEP_EVAL_STATE !== "1") diff --git a/packages/paperclip-runner/src/live/live-session.ts b/packages/paperclip-runner/src/live/live-session.ts index 4cefc77343c..dbb901b926f 100644 --- a/packages/paperclip-runner/src/live/live-session.ts +++ b/packages/paperclip-runner/src/live/live-session.ts @@ -265,6 +265,7 @@ export interface CapabilityLiveSessionSnapshot { runnerRestarts: number; runnerProcessPids: number[]; providerProcessPids: number[]; + sharedCodexServerPid: number | null; sameProviderThread: boolean; providerThreadDigest: string | null; callId: string; diff --git a/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts b/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts index 05d1a7c5d22..247a844ba4e 100644 --- a/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts +++ b/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts @@ -384,8 +384,20 @@ describe.sequential("Durable transport and recovery", () => { prompt: "Record one progress update.", model: "gpt-native-resume-fixture", runnerBinaryPath: durableRecoveryInternals.runnerBinary, + sharedProviderCommand: process.execPath, + sharedProviderSocketPath: resolve(root, "provider.sock"), + sharedProviderArgs: [ + providerFixture, + "server", + resolve(root, "provider.sock"), + resolve(root, "provider-state.json"), + ], providerCommand: process.execPath, - providerArgs: [providerFixture, resolve(root, "provider-state.json")], + providerArgs: [ + providerFixture, + "proxy", + resolve(root, "provider.sock"), + ], seed: { actors: [{ id: "actor-1", @@ -445,6 +457,7 @@ describe.sequential("Durable transport and recovery", () => { }); expect(new Set(proof.runnerProcessPids as number[]).size).toBe(2); expect(new Set(proof.providerProcessPids as number[]).size).toBe(2); + expect(proof.sharedCodexServerPid).toEqual(expect.any(Number)); expect(evidence.filter((entry) => { const data = entry.data as Record; return entry.kind === "tool_call" && data.callId === "call-native-resume"; diff --git a/packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs b/packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs index c7d00153c87..3acfb772568 100644 --- a/packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs +++ b/packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs @@ -1,175 +1,210 @@ #!/usr/bin/env node -import { readFileSync, writeFileSync } from "node:fs"; +import { readFileSync, unlinkSync, writeFileSync } from "node:fs"; +import { createConnection, createServer } from "node:net"; -const statePath = process.argv[2]; -if (!statePath) throw new Error("state path is required"); +const [mode, socketPath, statePath] = process.argv.slice(2); +if (!mode || !socketPath) throw new Error("mode and socket path are required"); -function load() { - try { - return JSON.parse(readFileSync(statePath, "utf8")); - } catch (error) { - if (error?.code !== "ENOENT") throw error; - return { - threadId: "thread-native-resume", - sessionId: "session-native-resume", - turnId: "turn-native-resume", - pendingTool: null, - completed: false, - }; - } -} - -let state = load(); -let buffer = ""; - -function save() { - writeFileSync(statePath, `${JSON.stringify(state)}\n`, { mode: 0o600 }); -} - -function send(value) { - process.stdout.write(`${JSON.stringify(value)}\n`); -} - -function sendPendingTool() { - if (!state.pendingTool) return; - send({ - id: state.pendingTool.requestId, - method: "item/tool/call", - params: { - threadId: state.threadId, - turnId: state.turnId, - callId: state.pendingTool.callId, - tool: "report_progress", - arguments: { - idempotencyKey: "native-resume-progress", - body: "One native resume effect.", - }, - }, +if (mode === "proxy") { + const socket = createConnection(socketPath); + socket.once("connect", () => { + process.stdin.pipe(socket); + socket.pipe(process.stdout); }); -} + socket.once("error", (error) => { + process.stderr.write(`${error.message}\n`); + process.exitCode = 1; + }); +} else if (mode === "server") { + if (!statePath) throw new Error("state path is required in server mode"); -function handleRequest(message) { - const { id, method } = message; - if (method === "initialize") { - send({ id, result: { user: { sessionId: state.sessionId } } }); - return; + function load() { + try { + return JSON.parse(readFileSync(statePath, "utf8")); + } catch (error) { + if (error?.code !== "ENOENT") throw error; + return { + threadId: "thread-native-resume", + sessionId: "session-native-resume", + turnId: "turn-native-resume", + pendingTool: null, + completed: false, + }; + } } - if (method === "thread/start" || method === "thread/resume") { - send({ - id, - result: { - model: "gpt-native-resume-fixture", - modelProvider: "openai-fixture", - thread: { id: state.threadId, sessionId: state.sessionId }, + + let state = load(); + + function save() { + writeFileSync(statePath, `${JSON.stringify(state)}\n`, { mode: 0o600 }); + } + + function send(socket, value) { + socket.write(`${JSON.stringify(value)}\n`); + } + + function sendPendingTool(socket) { + if (!state.pendingTool) return; + send(socket, { + id: state.pendingTool.requestId, + method: "item/tool/call", + params: { + threadId: state.threadId, + turnId: state.turnId, + callId: state.pendingTool.callId, + tool: "report_progress", + arguments: { + idempotencyKey: "native-resume-progress", + body: "One native resume effect.", + }, }, }); - if (method === "thread/resume") sendPendingTool(); - return; } - if (method === "thread/read") { - send({ - id, - result: { - thread: { - id: state.threadId, - sessionId: state.sessionId, - turns: [{ id: state.turnId, status: state.completed ? "completed" : "inProgress" }], - tokenUsage: { - total: { - inputTokens: 10, - cachedInputTokens: 0, - outputTokens: 2, - reasoningOutputTokens: 0, + + function handleRequest(socket, message) { + const { id, method } = message; + if (method === "initialize") { + send(socket, { id, result: { user: { sessionId: state.sessionId } } }); + return; + } + if (method === "thread/start" || method === "thread/resume") { + send(socket, { + id, + result: { + model: "gpt-native-resume-fixture", + modelProvider: "openai-fixture", + thread: { id: state.threadId, sessionId: state.sessionId }, + }, + }); + if (method === "thread/resume") sendPendingTool(socket); + return; + } + if (method === "thread/read") { + send(socket, { + id, + result: { + thread: { + id: state.threadId, + sessionId: state.sessionId, + turns: [{ + id: state.turnId, + status: state.completed ? "completed" : "inProgress", + }], + tokenUsage: { + total: { + inputTokens: 10, + cachedInputTokens: 0, + outputTokens: 2, + reasoningOutputTokens: 0, + }, }, }, }, + }); + return; + } + if (method === "turn/start") { + state.pendingTool = { + requestId: "rpc-native-resume", + callId: "call-native-resume", + }; + save(); + send(socket, { + id, + result: { turn: { id: state.turnId, status: "inProgress" } }, + }); + send(socket, { + method: "turn/started", + params: { + threadId: state.threadId, + turn: { id: state.turnId, status: "inProgress" }, + }, + }); + sendPendingTool(socket); + return; + } + send(socket, { + id, + error: { code: -32601, message: `unsupported method ${method}` }, + }); + } + + function handle(socket, message) { + if (message.method) { + if (message.id !== undefined) handleRequest(socket, message); + return; + } + if (!state.pendingTool || String(message.id) !== state.pendingTool.requestId) + return; + const toolResult = message.result; + state.pendingTool = null; + state.completed = true; + save(); + send(socket, { + method: "item/completed", + params: { + threadId: state.threadId, + turnId: state.turnId, + item: { + id: "message-native-resume", + type: "agentMessage", + text: `Tool result: ${JSON.stringify(toolResult)}`, + }, + }, + }); + send(socket, { + method: "thread/tokenUsage/updated", + params: { + threadId: state.threadId, + turnId: state.turnId, + tokenUsage: { + total: { + inputTokens: 10, + cachedInputTokens: 0, + outputTokens: 2, + reasoningTokens: 0, + }, + runDelta: { + requests: 1, + inputTokens: 10, + cacheReadTokens: 0, + outputTokens: 2, + reasoningTokens: 0, + providerCostUsd: 0, + }, + }, }, }); - return; - } - if (method === "turn/start") { - state.pendingTool = { - requestId: "rpc-native-resume", - callId: "call-native-resume", - }; - save(); - send({ id, result: { turn: { id: state.turnId, status: "inProgress" } } }); - send({ - method: "turn/started", + send(socket, { + method: "turn/completed", params: { threadId: state.threadId, - turn: { id: state.turnId, status: "inProgress" }, + turn: { id: state.turnId, status: "completed" }, }, }); - sendPendingTool(); - return; } - send({ id, error: { code: -32601, message: `unsupported method ${method}` } }); -} -function handle(message) { - if (message.method) { - if (message.id !== undefined) handleRequest(message); - return; + try { + unlinkSync(socketPath); + } catch (error) { + if (error?.code !== "ENOENT") throw error; } - if (!state.pendingTool || String(message.id) !== state.pendingTool.requestId) return; - const toolResult = message.result; - state.pendingTool = null; - state.completed = true; - save(); - send({ - method: "item/completed", - params: { - threadId: state.threadId, - turnId: state.turnId, - item: { - id: "message-native-resume", - type: "agentMessage", - text: `Tool result: ${JSON.stringify(toolResult)}`, - }, - }, - }); - send({ - method: "thread/tokenUsage/updated", - params: { - threadId: state.threadId, - turnId: state.turnId, - tokenUsage: { - total: { - inputTokens: 10, - cachedInputTokens: 0, - outputTokens: 2, - reasoningTokens: 0, - }, - runDelta: { - requests: 1, - inputTokens: 10, - cacheReadTokens: 0, - outputTokens: 2, - reasoningTokens: 0, - providerCostUsd: 0, - }, - }, - }, - }); - send({ - method: "turn/completed", - params: { - threadId: state.threadId, - turn: { id: state.turnId, status: "completed" }, - }, + const server = createServer((socket) => { + let buffer = ""; + socket.setEncoding("utf8"); + socket.on("data", (chunk) => { + buffer += chunk; + for (;;) { + const newline = buffer.indexOf("\n"); + if (newline < 0) break; + const line = buffer.slice(0, newline).trim(); + buffer = buffer.slice(newline + 1); + if (line) handle(socket, JSON.parse(line)); + } + }); }); + server.listen(socketPath); + process.once("SIGTERM", () => server.close()); +} else { + throw new Error(`unsupported mode ${mode}`); } - -process.stdin.setEncoding("utf8"); -process.stdin.resume(); -process.stdin.on("data", (chunk) => { - buffer += chunk; - for (;;) { - const newline = buffer.indexOf("\n"); - if (newline < 0) break; - const line = buffer.slice(0, newline).trim(); - buffer = buffer.slice(newline + 1); - if (line) handle(JSON.parse(line)); - } -});