fix(runner): speak WebSocket over Codex Unix sockets

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

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

@@ -1,6 +1,9 @@
 #!/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;
@@ -8,11 +11,131 @@ function socketArgument(argv: string[]): string {
   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", () => {
-    process.stdin.pipe(socket);
-    socket.pipe(process.stdout);
+    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`);

packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts

@@ -2596,8 +2596,14 @@ export async function runDurableEvalSession(
         await new Promise((resolveWait) => setTimeout(resolveWait, 20));
       }
       if (nativeResumeCall === null) {
+        handle.child.kill("SIGKILL");
+        const timedOutRunner = await handle.completion;
+        const diagnostics =
+          timedOutRunner.stderr.trim() ||
+          sharedCodexServerStderr.trim() ||
+          "no process diagnostics";
         throw new Error(
-          `native resume operation ${input.nativeResume.operationId} did not commit before timeout`,
+          `native resume operation ${input.nativeResume.operationId} did not commit before timeout: ${diagnostics}`,
         );
       }
       handle.child.kill("SIGKILL");

packages/paperclip-runner/src/mock-core/durable-recovery.test.ts

@@ -394,8 +394,13 @@ describe.sequential("Durable transport and recovery", () => {
         ],
         providerCommand: process.execPath,
         providerArgs: [
-          providerFixture,
-          "proxy",
+          fileURLToPath(
+            new URL(
+              "../../dist/cli/codex-app-server-unix-proxy.js",
+              import.meta.url,
+            ),
+          ),
+          "--socket",
           resolve(root, "provider.sock"),
         ],
         seed: {

packages/paperclip-runner/test/fixtures/fake-native-resume-codex-app-server.mjs

@@ -1,7 +1,28 @@
 #!/usr/bin/env node
+import { createHash } from "node:crypto";
 import { readFileSync, unlinkSync, writeFileSync } from "node:fs";
 import { createConnection, createServer } from "node:net";
 
+const websocketGuid = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
+
+function serverFrame(payload, opcode = 0x1) {
+  const body = Buffer.isBuffer(payload) ? payload : Buffer.from(payload);
+  const extended = body.length < 126 ? 0 : body.length <= 0xffff ? 2 : 8;
+  const frame = Buffer.alloc(2 + extended + body.length);
+  frame[0] = 0x80 | opcode;
+  frame[1] = body.length < 126 ? body.length : extended === 2 ? 126 : 127;
+  let offset = 2;
+  if (extended === 2) {
+    frame.writeUInt16BE(body.length, offset);
+    offset += 2;
+  } else if (extended === 8) {
+    frame.writeBigUInt64BE(BigInt(body.length), offset);
+    offset += 8;
+  }
+  body.copy(frame, offset);
+  return frame;
+}
+
 const [mode, socketPath, statePath] = process.argv.slice(2);
 if (!mode || !socketPath) throw new Error("mode and socket path are required");
 
@@ -40,7 +61,7 @@ if (mode === "proxy") {
   }
 
   function send(socket, value) {
-    socket.write(`${JSON.stringify(value)}\n`);
+    socket.write(serverFrame(JSON.stringify(value)));
   }
 
   function sendPendingTool(socket) {
@@ -190,16 +211,62 @@ if (mode === "proxy") {
     if (error?.code !== "ENOENT") throw error;
   }
   const server = createServer((socket) => {
-    let buffer = "";
-    socket.setEncoding("utf8");
+    let upgraded = false;
+    let buffer = Buffer.alloc(0);
     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));
+      buffer = Buffer.concat([buffer, chunk]);
+      if (!upgraded) {
+        const boundary = buffer.indexOf("\r\n\r\n");
+        if (boundary < 0) return;
+        const headers = buffer.subarray(0, boundary).toString("utf8");
+        const key = headers.match(/^sec-websocket-key:\s*(.+)$/im)?.[1]?.trim();
+        if (!key) throw new Error("WebSocket upgrade omitted Sec-WebSocket-Key");
+        const accept = createHash("sha1")
+          .update(`${key}${websocketGuid}`)
+          .digest("base64");
+        socket.write(
+          `HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`,
+        );
+        upgraded = true;
+        buffer = buffer.subarray(boundary + 4);
+      }
+
+      while (buffer.length >= 2) {
+        const first = buffer[0];
+        const second = buffer[1];
+        let length = second & 0x7f;
+        let offset = 2;
+        if (length === 126) {
+          if (buffer.length < 4) return;
+          length = buffer.readUInt16BE(2);
+          offset = 4;
+        } else if (length === 127) {
+          if (buffer.length < 10) return;
+          length = Number(buffer.readBigUInt64BE(2));
+          offset = 10;
+        }
+        if ((second & 0x80) === 0)
+          throw new Error("WebSocket client frame was not masked");
+        if (buffer.length < offset + 4 + length) return;
+        const mask = buffer.subarray(offset, offset + 4);
+        offset += 4;
+        const payload = Buffer.from(buffer.subarray(offset, offset + length));
+        buffer = buffer.subarray(offset + length);
+        for (let index = 0; index < payload.length; index += 1)
+          payload[index] ^= mask[index % 4];
+
+        const opcode = first & 0x0f;
+        if (opcode === 0x8) {
+          socket.end(serverFrame(payload, 0x8));
+          return;
+        }
+        if (opcode === 0x9) {
+          socket.write(serverFrame(payload, 0xa));
+          continue;
+        }
+        if (opcode !== 0x1)
+          throw new Error(`unsupported WebSocket opcode ${opcode}`);
+        handle(socket, JSON.parse(payload.toString("utf8")));
       }
     });
   });