From 457c9f8ddfce8d5a510f7690b847a5432a38e478 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 23 Aug 2026 20:29:01 +0000 Subject: [PATCH] fix(runner): speak WebSocket over Codex Unix sockets Co-Authored-By: Paperclip --- .../src/cli/codex-app-server-unix-proxy.ts | 127 +++++++++++++++++- .../durable-prp-control-plane.ts | 8 +- .../src/mock-core/durable-recovery.test.ts | 9 +- .../fake-native-resume-codex-app-server.mjs | 87 ++++++++++-- 4 files changed, 216 insertions(+), 15 deletions(-) diff --git a/packages/paperclip-runner/src/cli/codex-app-server-unix-proxy.ts b/packages/paperclip-runner/src/cli/codex-app-server-unix-proxy.ts index 1fc622f32c4..634d2508224 100644 --- a/packages/paperclip-runner/src/cli/codex-app-server-unix-proxy.ts +++ b/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`); 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 b28e3b5797a..303d9019ffe 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 @@ -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"); 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 247a844ba4e..e4b3c870dda 100644 --- a/packages/paperclip-runner/src/mock-core/durable-recovery.test.ts +++ b/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: { 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 3acfb772568..26cffb71c2f 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,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"))); } }); });