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 new file mode 100644 index 00000000000..064f4530a4e --- /dev/null +++ b/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; +} diff --git a/server/src/services/native-runtime/native-codex-runner.integration.test.ts b/server/src/services/native-runtime/native-codex-runner.integration.test.ts index 51be1e1fe11..6119139e30e 100644 --- a/server/src/services/native-runtime/native-codex-runner.integration.test.ts +++ b/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", diff --git a/server/src/services/native-runtime/native-codex-runner.ts b/server/src/services/native-runtime/native-codex-runner.ts index ac62e88a220..0b450ff9e36 100644 --- a/server/src/services/native-runtime/native-codex-runner.ts +++ b/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; + /** Internal qualification seam for recording the shared Codex server identity. */ + onProviderSpawn?: (meta: { pid: number; startedAt: string }) => Promise; }): Promise { 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 | null = null; + let sharedCodexServer: ChildProcess | null = null; + let sharedCodexExit: Promise | 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((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(); } }