fix(runner): preserve the Codex server across restarts

Run one local Codex app server outside Runner D and reconnect each Runner D generation through a private Unix WebSocket proxy. This keeps the production provider thread and pending JSON-RPC call alive when Runner D is killed.

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

packages/paperclip-runner/src/cli/codex-app-server-unix-proxy.ts

@@ -0,0 +1,147 @@
+#!/usr/bin/env node
+import { createHash, randomBytes } from "node:crypto";
+import { createConnection } from "node:net";
+
+const websocketGuid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
+
+function socketArgument(argv: string[]): string {
+  const index = argv.indexOf("--socket");
+  const value = index >= 0 ? argv[index + 1] : undefined;
+  if (!value) throw new Error("--socket requires a Unix socket path");
+  return value;
+}
+
+function clientFrame(payload: Buffer, opcode = 0x1): Buffer {
+  const length = payload.length;
+  const extended = length < 126 ? 0 : length <= 0xffff ? 2 : 8;
+  const frame = Buffer.alloc(2 + extended + 4 + length);
+  frame[0] = 0x80 | opcode;
+  frame[1] = 0x80 | (length < 126 ? length : extended === 2 ? 126 : 127);
+  let offset = 2;
+  if (extended === 2) {
+    frame.writeUInt16BE(length, offset);
+    offset += 2;
+  } else if (extended === 8) {
+    frame.writeBigUInt64BE(BigInt(length), offset);
+    offset += 8;
+  }
+  const mask = randomBytes(4);
+  mask.copy(frame, offset);
+  offset += 4;
+  for (let index = 0; index < length; index += 1) {
+    frame[offset + index] = payload[index]! ^ mask[index % 4]!;
+  }
+  return frame;
+}
+
+try {
+  const socket = createConnection(socketArgument(process.argv.slice(2)));
+  const websocketKey = randomBytes(16).toString("base64");
+  const expectedAccept = createHash("sha1")
+    .update(`${websocketKey}${websocketGuid}`)
+    .digest("base64");
+  let connected = false;
+  let networkBuffer = Buffer.alloc(0);
+  let inputBuffer = "";
+  let fragmentedPayloads: Buffer[] = [];
+
+  function emitMessage(payload: Buffer): void {
+    process.stdout.write(payload);
+    process.stdout.write("\n");
+  }
+
+  function readFrames(): void {
+    while (networkBuffer.length >= 2) {
+      const first = networkBuffer[0]!;
+      const second = networkBuffer[1]!;
+      const masked = (second & 0x80) !== 0;
+      let length = second & 0x7f;
+      let offset = 2;
+      if (length === 126) {
+        if (networkBuffer.length < 4) return;
+        length = networkBuffer.readUInt16BE(2);
+        offset = 4;
+      } else if (length === 127) {
+        if (networkBuffer.length < 10) return;
+        const wideLength = networkBuffer.readBigUInt64BE(2);
+        if (wideLength > BigInt(Number.MAX_SAFE_INTEGER)) {
+          throw new Error("Codex app-server WebSocket frame is too large");
+        }
+        length = Number(wideLength);
+        offset = 10;
+      }
+      const maskLength = masked ? 4 : 0;
+      if (networkBuffer.length < offset + maskLength + length) return;
+      const mask = masked ? networkBuffer.subarray(offset, offset + 4) : null;
+      offset += maskLength;
+      const payload = Buffer.from(networkBuffer.subarray(offset, offset + length));
+      networkBuffer = networkBuffer.subarray(offset + length);
+      if (mask) {
+        for (let index = 0; index < payload.length; index += 1) {
+          payload[index] = payload[index]! ^ mask[index % 4]!;
+        }
+      }
+
+      const final = (first & 0x80) !== 0;
+      const opcode = first & 0x0f;
+      if (opcode === 0x8) {
+        socket.end(clientFrame(payload, 0x8));
+        return;
+      }
+      if (opcode === 0x9) {
+        socket.write(clientFrame(payload, 0xa));
+        continue;
+      }
+      if (opcode === 0x1 || opcode === 0x2 || opcode === 0x0) {
+        fragmentedPayloads.push(payload);
+        if (final) {
+          emitMessage(Buffer.concat(fragmentedPayloads));
+          fragmentedPayloads = [];
+        }
+      }
+    }
+  }
+
+  socket.once("connect", () => {
+    socket.write(
+      `GET / HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: ${websocketKey}\r\nSec-WebSocket-Version: 13\r\n\r\n`,
+    );
+  });
+  socket.on("data", (chunk: Buffer) => {
+    networkBuffer = Buffer.concat([networkBuffer, chunk]);
+    if (!connected) {
+      const boundary = networkBuffer.indexOf("\r\n\r\n");
+      if (boundary < 0) return;
+      const headers = networkBuffer.subarray(0, boundary).toString("utf8");
+      networkBuffer = networkBuffer.subarray(boundary + 4);
+      if (!headers.startsWith("HTTP/1.1 101 ")) {
+        throw new Error("Codex app-server rejected the WebSocket upgrade");
+      }
+      const accept = headers.match(/^sec-websocket-accept:\s*(.+)$/im)?.[1]?.trim();
+      if (accept !== expectedAccept) {
+        throw new Error("Codex app-server returned an invalid WebSocket accept value");
+      }
+      connected = true;
+      process.stdin.setEncoding("utf8");
+      process.stdin.on("data", (chunk: string) => {
+        inputBuffer += chunk;
+        for (;;) {
+          const newline = inputBuffer.indexOf("\n");
+          if (newline < 0) break;
+          const line = inputBuffer.slice(0, newline).trim();
+          inputBuffer = inputBuffer.slice(newline + 1);
+          if (line) socket.write(clientFrame(Buffer.from(line)));
+        }
+      });
+      process.stdin.resume();
+    }
+    readFrames();
+  });
+  socket.once("error", (error) => {
+    process.stderr.write(`Codex app-server socket error: ${error.message}\n`);
+    process.exitCode = 1;
+  });
+} catch (error) {
+  process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`);
+  process.exitCode = 1;
+}

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

@@ -180,6 +180,7 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
     });
     const logs: string[] = [];
     const runnerPids: number[] = [];
+    const providerPids: number[] = [];
     const liveCodex = process.env.PAPERCLIP_LIVE_CODEX_NATIVE_RESUME === "1";
     let semanticCallId: string | null = null;
     let resolveRestart!: () => void;
@@ -239,6 +240,10 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
         if (liveCodex) console.log("NATIVE_CODEX_RESUME_RUNNER", pid);
         if (runnerPids.length === 2) resolveRestart();
       },
+      onProviderSpawn: async ({ pid }) => {
+        providerPids.push(pid);
+        if (liveCodex) console.log("NATIVE_CODEX_RESUME_PROVIDER", pid);
+      },
     });
     const result = await execute.catch((error) => {
       throw new Error(
@@ -335,6 +340,7 @@ describeEmbeddedPostgres("native Codex server vertical slice", () => {
         providerSessionId: result.sessionParams?.sessionId ?? null,
         semanticCallId,
         runnerPids,
+        providerPids,
         semanticInputCount: semanticInputs.length,
         semanticReconciledCount: nativeEvents.filter(
           (event) => event.eventType === "semantic_tool.reconciled",

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

@@ -1,6 +1,6 @@
 import { spawn, type ChildProcess } from "node:child_process";
 import { createHash } from "node:crypto";
-import { accessSync, chmodSync, constants, mkdirSync, readFileSync } from "node:fs";
+import { accessSync, chmodSync, constants, existsSync, mkdirSync, readFileSync } from "node:fs";
 import { dirname, isAbsolute, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 
@@ -47,6 +47,25 @@ export function resolvePaperclipRunnerBinary(
   );
 }
 
+function resolveCodexUnixProxy(): string {
+  const candidates = [
+    resolve(moduleDirectory, "../../vendor/paperclip-runner/cli/codex-app-server-unix-proxy.js"),
+    resolve(
+      moduleDirectory,
+      "../../../../packages/paperclip-runner/dist/cli/codex-app-server-unix-proxy.js",
+    ),
+  ];
+  for (const candidate of candidates) {
+    try {
+      accessSync(candidate, constants.R_OK);
+      return candidate;
+    } catch {
+      // Continue through packaged and workspace locations.
+    }
+  }
+  throw new Error("paperclip_runner_codex_proxy_missing: build @paperclipai/paperclip-runner");
+}
+
 export function buildNativeRunnerArguments(input: {
   connectUrl: string;
   stateDirectory: string;
@@ -169,6 +188,8 @@ export async function executeNativeCodexRunner(input: {
     processGroupId: number | null;
     startedAt: string;
   }) => Promise<void>;
+  /** Internal qualification seam for recording the shared Codex server identity. */
+  onProviderSpawn?: (meta: { pid: number; startedAt: string }) => Promise<void>;
 }): Promise<AdapterExecutionResult> {
   const binary = input.runnerBinary ?? resolvePaperclipRunnerBinary();
   const runnerDigest = `sha256:${createHash("sha256").update(readFileSync(binary)).digest("hex")}`;
@@ -180,6 +201,11 @@ export async function executeNativeCodexRunner(input: {
   privateDirectory(resolve(runtimeRoot, "control-plane"));
   privateDirectory(resolve(runtimeRoot, "runner"));
   privateDirectory(runnerStateDirectory);
+  const sharedCodexSocket = resolve(runnerStateDirectory, "codex-app-server.sock");
+  const useSharedCodexServer = input.providerLaunch === undefined;
+  const providerCommand = input.providerLaunch?.command ?? process.execPath;
+  const providerArgs = input.providerLaunch?.args
+    ?? [resolveCodexUnixProxy(), "--socket", sharedCodexSocket];
 
   const prepared = await runnerPrpCoordinator(input.db, {
     stateRoot: resolve(runtimeRoot, "control-plane"),
@@ -203,8 +229,8 @@ export async function executeNativeCodexRunner(input: {
       provider: "codex",
       driver: "codex_app_server",
       providerVersion: input.providerLaunch?.providerVersion ?? "codex-app-server-v1",
-      command: input.providerLaunch?.command ?? "codex",
-      args: input.providerLaunch?.args ?? ["app-server"],
+      command: providerCommand,
+      args: providerArgs,
       cwd: input.cwd,
       ...(input.model ? { model: input.model } : {}),
       ...(input.resumeProviderSessionId
@@ -221,7 +247,52 @@ export async function executeNativeCodexRunner(input: {
 
   let activeChild: ChildProcess | null = null;
   let activeExit: Promise<unknown> | null = null;
+  let sharedCodexServer: ChildProcess | null = null;
+  let sharedCodexExit: Promise<unknown> | null = null;
   try {
+    if (useSharedCodexServer) {
+      let startupError: Error | null = null;
+      let stderr = "";
+      sharedCodexServer = spawn(
+        "codex",
+        ["app-server", "--listen", `unix://${sharedCodexSocket}`],
+        {
+          cwd: input.cwd,
+          detached: process.platform !== "win32",
+          env: { ...process.env, ...input.environment },
+          stdio: ["ignore", "ignore", "pipe"],
+        },
+      );
+      sharedCodexExit = waitForExit(sharedCodexServer);
+      sharedCodexServer.once("error", (error) => {
+        startupError = error;
+      });
+      sharedCodexServer.stderr?.setEncoding("utf8").on("data", (chunk: string) => {
+        stderr = `${stderr}${chunk}`.slice(-16_384);
+      });
+      if (!sharedCodexServer.pid) throw new Error("codex_app_server_process_not_started");
+      await input.onProviderSpawn?.({
+        pid: sharedCodexServer.pid,
+        startedAt: new Date().toISOString(),
+      });
+      const startupDeadline = Date.now() + 10_000;
+      while (
+        !existsSync(sharedCodexSocket)
+        && startupError === null
+        && sharedCodexServer.exitCode === null
+        && Date.now() < startupDeadline
+      ) {
+        await new Promise<void>((resolveWait) => {
+          const timer = setTimeout(resolveWait, 20);
+          timer.unref();
+        });
+      }
+      if (!existsSync(sharedCodexSocket)) {
+        throw new Error(
+          `codex_app_server_socket_missing: ${(startupError as Error | null)?.message ?? stderr.trim()}`,
+        );
+      }
+    }
     const terminal = prepared.waitForTerminal(input.timeoutMs);
     let bootstrapTicket = prepared.bootstrapTicket;
     let restartCount = 0;
@@ -326,6 +397,9 @@ export async function executeNativeCodexRunner(input: {
     if (activeChild && activeExit) {
       await stopChild(activeChild, activeExit).catch(() => undefined);
     }
+    if (sharedCodexServer && sharedCodexExit) {
+      await stopChild(sharedCodexServer, sharedCodexExit).catch(() => undefined);
+    }
     await prepared.release();
   }
 }