fix(runner): rejoin a shared Codex app-server

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

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<void>((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")

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;

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<string, unknown>;
         return entry.kind === "tool_call" && data.callId === "call-native-resume";

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));
-  }
-});