From 83243d4b5dedd7d96ed88007576c4480a4c68290 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 30 Aug 2026 00:07:50 -0500 Subject: [PATCH] feat(runner): add bounded Codex app-server transport (#12362) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The first production runner provider is Codex > - Codex app-server communicates with JSON-RPC over a child process > - The process boundary needs explicit resource, environment, and diagnostic limits > - This pull request adds only the bounded transport layer > - A later pull request will add the Codex event-mapping driver > - The benefit is a small reviewable process boundary before provider behavior is added ## Linked Issues or Issue Description **Subsystem affected** `packages/paperclip-runner` Codex transport. **Problem or motivation** The runner needs a safe way to start and communicate with Codex app-server. Unbounded lines, queued notifications, pending requests, and inherited credentials would make that boundary unsafe. **Proposed solution** Add a JSON-RPC process transport with explicit byte and count limits, sanitized environment construction, process-group metadata, server-request handling, redacted diagnostics, and deterministic closure. **Alternatives considered** Adding transport and the full Codex driver in one change would create a much larger review surface. This pull request isolates the child-process boundary. **Roadmap alignment** This supports the Codex-first experimental runner. It does not enable the runner adapter. ## What Changed - Added bounded JSON-RPC request and notification queues. - Added malformed-message and oversized-line fail-closed behavior. - Added sanitized Codex environment construction. - Added process-group tracking and deterministic shutdown. - Added redacted diagnostics and server-request replies. ## Verification - `pnpm --filter @paperclipai/paperclip-runner test:typescript` - `pnpm -r typecheck` - `pnpm build` - The focused transport test has 6 passing cases. ## Risks The main risk is child-process cleanup. Tests cover closure, pending request rejection, queue bounds, process metadata, and malformed input. ## Model Used OpenAI Codex with GPT-5.6 and repository tool use. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --- .../codex/app-server-transport.test.ts | 248 ++++++ .../src/drivers/codex/app-server-transport.ts | 781 ++++++++++++++++++ 2 files changed, 1029 insertions(+) create mode 100644 packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts create mode 100644 packages/paperclip-runner/src/drivers/codex/app-server-transport.ts diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts new file mode 100644 index 0000000000..5ca2301d33 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.test.ts @@ -0,0 +1,248 @@ +import { ChildProcess } from "node:child_process"; +import { describe, expect, it, vi } from "vitest"; + +import { ProcessCodexAppServerTransport, redactCodexDiagnostic } from "./app-server-transport.js"; + +function nodeTransport( + source: string, + options: ConstructorParameters[0] = {}, +) { + return new ProcessCodexAppServerTransport({ + ...options, + command: process.execPath, + args: ["-e", source], + environment: { PATH: process.env.PATH }, + }); +} + +describe("Codex app-server transport limits", () => { + it("redacts real Basic credentials without corrupting ordinary question copy", () => { + expect(redactCodexDiagnostic("Authorization: Basic dXNlcjpwYXNz")) + .toBe("Authorization: Basic [REDACTED]"); + expect(redactCodexDiagnostic("Basic API foundation")) + .toBe("Basic API foundation"); + }); + + it("reports restart-safe process-group ownership", async () => { + const transport = nodeTransport("process.stdin.resume()", { processGroup: true }); + const info = transport.processInfo(); + + expect(info.pid).toBeGreaterThan(0); + expect(info.processGroupId).toBe(process.platform === "win32" ? null : info.pid); + expect(new Date(info.startedAt).toISOString()).toBe(info.startedAt); + + await transport.close(); + }); + + it("installs cleanup handlers before synchronously reporting process ownership", async () => { + const originalSpawn = ChildProcess.prototype.spawn; + let spawnedProcess: ChildProcess | undefined; + const spawnSpy = vi + .spyOn(ChildProcess.prototype, "spawn") + .mockImplementation(function (this: ChildProcess, ...args) { + spawnedProcess = this; + return originalSpawn.apply(this, args); + }); + let transport: ProcessCodexAppServerTransport | undefined; + let observedInitialProcess = false; + + try { + transport = nodeTransport("process.stdin.resume()", { + onProcess: (info) => { + if (info.exited) return; + observedInitialProcess = true; + expect(spawnedProcess?.pid).toBe(info.pid); + expect(spawnedProcess?.listenerCount("error")).toBeGreaterThan(0); + expect(spawnedProcess?.listenerCount("exit")).toBeGreaterThan(0); + expect( + spawnedProcess?.stdout?.listenerCount("data"), + ).toBeGreaterThan(0); + expect( + spawnedProcess?.stdout?.listenerCount("end"), + ).toBeGreaterThan(0); + expect( + spawnedProcess?.stderr?.listenerCount("data"), + ).toBeGreaterThan(0); + expect( + spawnedProcess?.stdin?.listenerCount("error"), + ).toBeGreaterThan(0); + spawnedProcess?.emit( + "error", + new Error("synchronous process callback cleanup"), + ); + }, + }); + expect(observedInitialProcess).toBe(true); + await expect( + transport.request("after-callback-cleanup", {}), + ).rejects.toThrow("codex app-server transport is closed"); + } finally { + spawnSpy.mockRestore(); + await transport?.close(); + } + }); + + it("rejects an oversized line before buffering the complete hostile payload", async () => { + const diagnostics: string[] = []; + const transport = nodeTransport( + "setTimeout(() => process.stdout.write('x'.repeat(1048576)), 50); setInterval(() => {}, 1000)", + { maxLineBytes: 128, onDiagnostic: (message) => diagnostics.push(message) }, + ); + await expect(transport.notifications()[Symbol.asyncIterator]().next()).rejects.toThrow( + "codex app-server line exceeded 128 bytes", + ); + expect(diagnostics).toEqual(["codex app-server line exceeded 128 bytes"]); + await transport.close(); + }); + + it("bounds pending requests", async () => { + const transport = nodeTransport("process.stdin.resume()", { maxPendingRequests: 1 }); + const first = transport.request("first", {}); + await expect(transport.request("second", {})).rejects.toThrow( + "codex app-server pending request limit 1 exceeded", + ); + const firstRejected = expect(first).rejects.toThrow("codex app-server transport closed"); + await transport.close(); + await firstRejected; + }); + + it("fails closed when queued notifications exceed their count bound", async () => { + let transport: ProcessCodexAppServerTransport; + const diagnostic = new Promise((resolve) => { + const lines = [1, 2].map((id) => JSON.stringify({ + method: "item/completed", + params: { threadId: "thread-1", turnId: "turn-1", item: { id } }, + })).join("\n"); + transport = nodeTransport( + `setTimeout(() => process.stdout.write(${JSON.stringify(`${lines}\n`)}), 50); setInterval(() => {}, 1000)`, + { + maxQueuedNotifications: 1, + maxQueuedNotificationBytes: 1024, + onDiagnostic: (message) => resolve(message), + }, + ); + }); + await expect(diagnostic).resolves.toBe("codex app-server notification queue limit exceeded"); + await transport!.close(); + }); + + it("rejects malformed JSON-RPC messages", async () => { + const transport = nodeTransport( + "setTimeout(() => process.stdout.write('{not-json}\\n'), 50); setInterval(() => {}, 1000)", + ); + await expect(transport.request("pending", {})).rejects.toThrow( + "codex app-server emitted malformed JSON", + ); + await transport.close(); + }); + + it("fails pending and future requests when stdout closes cleanly", async () => { + const transport = nodeTransport( + "process.stdin.resume(); process.stdout.end(); setInterval(() => {}, 1000)", + ); + const pending = transport.request("pending", {}); + await expect(pending).rejects.toThrow( + "codex app-server stdout ended before transport closure", + ); + await expect(transport.request("after-stdout-end", {})).rejects.toThrow( + "codex app-server transport is closed", + ); + await transport.close(); + }); + + it("fails closed when stdout is destroyed without a usable response channel", async () => { + const transport = nodeTransport( + 'process.stdin.resume(); require("node:fs").closeSync(1); setInterval(() => {}, 1000)', + ); + await expect(transport.request("pending", {})).rejects.toThrow(/stdout (ended|closed)/u); + await expect(transport.request("after-stdout-close", {})).rejects.toThrow( + "codex app-server transport is closed", + ); + await transport.close(); + }); + + it("fails closed when outbound buffering exceeds its bound", async () => { + const diagnostics: string[] = []; + const transport = nodeTransport("process.stdin.pause(); setInterval(() => {}, 1000)", { + maxLineBytes: 1_024, + maxBufferedOutputBytes: 64, + onDiagnostic: (message) => diagnostics.push(message), + }); + + expect(() => transport.notify("large", { value: "x".repeat(100) })).toThrow( + "outbound codex JSON-RPC buffer exceeded 64 bytes", + ); + expect(diagnostics).toContain( + "outbound codex JSON-RPC buffer exceeded 64 bytes", + ); + await expect(transport.request("after-close", {})).rejects.toThrow( + "codex app-server transport is closed", + ); + await transport.close(); + }); + + it("routes an oversized server response through deterministic closure", async () => { + let transport: ProcessCodexAppServerTransport | undefined; + const diagnostic = new Promise((resolve) => { + transport = nodeTransport( + `setTimeout(() => process.stdout.write(JSON.stringify({ id: "server-1", method: "tool/call", params: {} }) + "\\n"), 20); setInterval(() => {}, 1000)`, + { + maxLineBytes: 128, + onDiagnostic: (message) => resolve(message), + }, + ); + transport.setServerRequestHandler(async () => ({ value: "x".repeat(256) })); + }); + + await expect(diagnostic).resolves.toBe( + "outbound codex JSON-RPC line exceeded 128 bytes", + ); + await transport?.close(); + }); + + it("turns a synchronous server handler throw into a JSON-RPC error", async () => { + const transport = nodeTransport(` + process.stdin.setEncoding("utf8"); + process.stdin.once("data", (chunk) => { + const response = JSON.parse(chunk.trim()); + process.stdout.write(JSON.stringify({ method: "observed", params: response }) + "\\n"); + }); + setTimeout(() => process.stdout.write(JSON.stringify({ id: "server-1", method: "tool/call", params: {} }) + "\\n"), 20); + setInterval(() => {}, 1000); + `); + transport.setServerRequestHandler((): Promise> => { + throw new Error("synchronous handler failure"); + }); + + await expect( + transport.notifications()[Symbol.asyncIterator]().next(), + ).resolves.toMatchObject({ + done: false, + value: { + method: "observed", + params: { + id: "server-1", + error: { code: -32_000, message: "Error: synchronous handler failure" }, + }, + }, + }); + await transport.close(); + }); + + it("routes an asynchronous child-stdin failure through deterministic closure", async () => { + const diagnostics: string[] = []; + const transport = nodeTransport( + `require("node:fs").closeSync(0); process.stdout.write(JSON.stringify({ method: "ready", params: {} }) + "\\n"); setInterval(() => {}, 1000)`, + { onDiagnostic: (message) => diagnostics.push(message) }, + ); + const notifications = transport.notifications()[Symbol.asyncIterator](); + await expect(notifications.next()).resolves.toMatchObject({ + done: false, + value: { method: "ready" }, + }); + + await expect(transport.request("after-stdin-close", {})).rejects.toThrow(); + expect(diagnostics.some((message) => /EPIPE|closed|write/u.test(message))).toBe(true); + await transport.close(); + }); +}); diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts new file mode 100644 index 0000000000..e331a60399 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -0,0 +1,781 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import type { HarnessRuntimeRequestResolution } from "../../contracts/harness-driver.js"; + +export interface CodexRpcNotification { + method: string; + params: Record; + /** Internal-only correlation retained outside provider params and canonical PRP payloads. */ + paperclipTrace?: { + sourceEventId: string; + sourceEventType: string; + }; +} + +export interface CodexTraceInterpretation { + sourceEventId: string; + sourceEventType: string; + providerMethod: string; + disposition: "mapped" | "ignored" | "rejected"; + emittedEventIds: string[]; + reason: string; +} + +export interface CodexRpcServerRequest extends CodexRpcNotification { + id: string | number; +} + +export type CodexServerRequestHandler = ( + request: CodexRpcServerRequest, +) => Promise>; + +export interface CodexAppServerTransport { + request( + method: string, + params: Record, + ): Promise>; + notify(method: string, params?: Record): void; + notifications(): AsyncIterable; + setServerRequestHandler(handler: CodexServerRequestHandler): void; + /** + * Optional provider-neutral resolution path used when a transport has + * already normalized a native server request behind another PRP boundary. + */ + resolveRuntimeRequest?(input: { + requestId: string; + turnId: string; + resolution: HarnessRuntimeRequestResolution; + }): Promise; + close(): Promise; + processInfo?(): CodexTransportProcessInfo; + attachRun?(input: { + runId: string; + turnId: string; + itemId: string; + }): Promise; + /** Records post-rehydration driver mapping without affecting run authority. */ + recordTraceInterpretation?(input: CodexTraceInterpretation): void; +} + +export interface CodexTransportProcessInfo { + pid: number | null; + processGroupId: number | null; + startedAt: string; + exited: boolean; + exitCode: number | null; + signal: NodeJS.Signals | null; +} + +class BoundedAsyncQueue implements AsyncIterable { + #values: Array<{ value: T; bytes: number }> = []; + #waiters: Array<{ + resolve: (value: IteratorResult) => void; + reject: (error: Error) => void; + }> = []; + #queuedBytes = 0; + #closed = false; + #error: Error | null = null; + + constructor( + private readonly maxValues: number, + private readonly maxBytes: number, + ) {} + + push(value: T, bytes: number): boolean { + if (this.#closed) return false; + const waiter = this.#waiters.shift(); + if (waiter !== undefined) { + waiter.resolve({ value, done: false }); + return true; + } + if ( + this.#values.length >= this.maxValues || + this.#queuedBytes + bytes > this.maxBytes + ) { + return false; + } + this.#values.push({ value, bytes }); + this.#queuedBytes += bytes; + return true; + } + + close(error?: Error): void { + if (this.#closed) return; + this.#closed = true; + this.#values = []; + this.#queuedBytes = 0; + this.#error = error ?? null; + for (const waiter of this.#waiters.splice(0)) { + if (error === undefined) waiter.resolve({ value: undefined, done: true }); + else waiter.reject(error); + } + } + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: async () => { + const queued = this.#values.shift(); + if (queued !== undefined) { + this.#queuedBytes -= queued.bytes; + return { value: queued.value, done: false }; + } + if (this.#error !== null) return Promise.reject(this.#error); + if (this.#closed) return { value: undefined, done: true }; + return new Promise>((resolve, reject) => { + this.#waiters.push({ resolve, reject }); + }); + }, + }; + } +} + +class BoundedLineDecoder { + #buffer = Buffer.alloc(0); + #closed = false; + + constructor( + private readonly maxLineBytes: number, + private readonly onLine: (line: string, bytes: number) => void, + private readonly onError: (error: Error) => void, + ) {} + + write(chunk: Buffer | string): void { + if (this.#closed) return; + const incoming = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + let offset = 0; + while (offset < incoming.length) { + const newline = incoming.indexOf(0x0a, offset); + const end = newline < 0 ? incoming.length : newline; + const segment = incoming.subarray(offset, end); + if (this.#buffer.length + segment.length > this.maxLineBytes) { + this.#fail( + new Error( + `codex app-server line exceeded ${this.maxLineBytes} bytes`, + ), + ); + return; + } + if (segment.length > 0) + this.#buffer = Buffer.concat([this.#buffer, segment]); + if (newline < 0) return; + const line = this.#buffer; + this.#buffer = Buffer.alloc(0); + this.onLine(line.toString("utf8"), line.length); + if (this.#closed) return; + offset = newline + 1; + } + } + + end(): void { + if (this.#closed) return; + if (this.#buffer.length > 0) { + this.#fail( + new Error("codex app-server ended with an unterminated JSON-RPC line"), + ); + } + this.#closed = true; + } + + close(): void { + this.#closed = true; + this.#buffer = Buffer.alloc(0); + } + + #fail(error: Error): void { + this.close(); + this.onError(error); + } +} + +const SAFE_ENVIRONMENT_KEYS = [ + "ALL_PROXY", + "CODEX_HOME", + "HOME", + "HTTP_PROXY", + "HTTPS_PROXY", + "LANG", + "LC_ALL", + "NO_PROXY", + "NODE_EXTRA_CA_CERTS", + "PATH", + "PATHEXT", + "SSL_CERT_FILE", + "SystemRoot", + "TEMP", + "TMP", + "TMPDIR", + "WINDIR", +] as const; + +/** + * Environment for the trusted app-server process itself. HOME and CODEX_HOME + * are retained so Codex can authenticate; model-issued commands receive a + * separate empty-by-default environment and filesystem permission profile. + */ +export function createSanitizedCodexEnvironment( + source: NodeJS.ProcessEnv = process.env, +): NodeJS.ProcessEnv { + const environment: NodeJS.ProcessEnv = {}; + for (const key of SAFE_ENVIRONMENT_KEYS) { + const value = source[key]; + if (value === undefined) continue; + if (key.includes("PROXY") && proxyContainsCredentials(value)) continue; + environment[key] = value; + } + return environment; +} + +export function sanitizedEnvironmentKeys( + source: NodeJS.ProcessEnv = process.env, +): string[] { + return Object.keys(createSanitizedCodexEnvironment(source)).sort(); +} + +export function redactCodexDiagnostic(message: string): string { + return message + .replaceAll(/\u001b\[[0-?]*[ -/]*[@-~]/g, "") + .replace(/Bearer\s+[A-Za-z0-9._~+\/-]+/gi, "Bearer [REDACTED]") + .replace(/Basic\s+([A-Za-z0-9+/=]+)/gi, (match, encoded: string) => { + try { + // Only redact an actual RFC 7617 credential. Treating every word after + // “Basic” as base64 corrupted ordinary question copy such as + // “Basic API” before it entered the Paperclip protocol. + const decoded = Buffer.from(encoded, "base64").toString("utf8"); + return decoded.includes(":") ? "Basic [REDACTED]" : match; + } catch { + return match; + } + }) + .replace(/([a-z][a-z0-9+.-]*:\/\/)[^\s/@:]+:[^\s/@]+@/gi, "$1[REDACTED]@") + .replace( + /([?&](?:api[_-]?key|token|secret|password)=)[^&#\s]+/gi, + "$1[REDACTED]", + ) + .replace( + /(["'](?:api[_-]?key|token|secret|password|authorization)["']\s*:\s*["'])[^"']+/gi, + "$1[REDACTED]", + ) + .replace( + /(api[_-]?key|token|secret|password)\s*[=:]\s*[^\s,;]+/gi, + "$1=[REDACTED]", + ) + .replace( + /(PAPERCLIP_API_KEY|OPENAI_API_KEY|OPENROUTER_API_KEY)=[^\s]+/g, + "$1=[REDACTED]", + ); +} + +function proxyContainsCredentials(value: string): boolean { + try { + const url = new URL(value); + return url.username.length > 0 || url.password.length > 0; + } catch { + return true; + } +} + +export const CODEX_METHOD_NOT_FOUND = -32_601; +export const CODEX_INVALID_REQUEST = -32_600; + +/** + * Codes where the provider refused the call itself rather than failing while + * serving it: the method is absent (-32601), or the request was rejected + * outright, which is how the Codex app-server reports a feature it has + * switched off (-32600 `goals feature is disabled`). Server-error codes + * (-32603 and the -32000..-32099 implementation range) and transport failures + * describe one attempt, not what the build supports. + */ +const METHOD_UNAVAILABLE_CODES: readonly number[] = [ + CODEX_METHOD_NOT_FOUND, + CODEX_INVALID_REQUEST, +]; + +/** + * A provider error response, carrying its JSON-RPC code. Callers need the code + * to tell "this provider build will not serve the method" apart from "the call + * failed this time", which are different facts about a capability. + */ +export class CodexRpcError extends Error { + readonly code: number | null; + + constructor(message: string, code: number | null) { + super(message); + this.name = "CodexRpcError"; + this.code = code; + } + + get methodUnavailable(): boolean { + return this.code !== null && METHOD_UNAVAILABLE_CODES.includes(this.code); + } +} + +/** + * True only when the provider answered and its answer denies the method. A + * plain `Error` here means the call never got an answer — a transport or + * protocol failure — which must not be read as a missing capability. + */ +export function isCodexMethodUnavailable(error: unknown): boolean { + return error instanceof CodexRpcError && error.methodUnavailable; +} + +function rpcErrorCode(value: unknown): number | null { + if (typeof value !== "object" || value === null) return null; + const code = (value as { code?: unknown }).code; + return typeof code === "number" && Number.isFinite(code) ? code : null; +} + +interface PendingRequest { + resolve: (value: Record) => void; + reject: (error: Error) => void; +} + +export interface ProcessCodexTransportOptions { + command?: string; + args?: string[]; + environment?: NodeJS.ProcessEnv; + onDiagnostic?: (message: string) => void; + maxLineBytes?: number; + maxPendingRequests?: number; + maxQueuedNotifications?: number; + maxQueuedNotificationBytes?: number; + maxBufferedOutputBytes?: number; + maxInflightServerRequests?: number; + /** Starts a dedicated process group so runnerd and its Codex child are reaped together. */ + processGroup?: boolean; + closeGraceMs?: number; + onProcess?: (info: CodexTransportProcessInfo) => void; +} + +const DEFAULT_MAX_LINE_BYTES = 1024 * 1024; +const DEFAULT_MAX_PENDING_REQUESTS = 64; +const DEFAULT_MAX_QUEUED_NOTIFICATIONS = 256; +const DEFAULT_MAX_QUEUED_NOTIFICATION_BYTES = 4 * 1024 * 1024; +const DEFAULT_MAX_BUFFERED_OUTPUT_BYTES = 4 * 1024 * 1024; +const DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS = 32; +const MAX_DIAGNOSTIC_LINE_BYTES = 16 * 1024; + +/** Local stdio JSON-RPC transport. App-server never becomes a WAN endpoint. */ +export class ProcessCodexAppServerTransport implements CodexAppServerTransport { + #process: ChildProcessWithoutNullStreams; + #notifications: BoundedAsyncQueue; + #pending = new Map(); + #nextId = 1; + #serverRequestHandler: CodexServerRequestHandler = async () => ({ + success: false, + contentItems: [{ type: "input_text", text: "Unsupported client request." }], + }); + #inflightServerRequests = 0; + #closed = false; + #exited = false; + #exitCode: number | null = null; + #exitSignal: NodeJS.Signals | null = null; + #resolveExit!: () => void; + readonly #exitPromise: Promise; + readonly #processGroup: boolean; + readonly #startedAt: string; + readonly #closeGraceMs: number; + readonly #onProcess?: (info: CodexTransportProcessInfo) => void; + #onDiagnostic?: (message: string) => void; + readonly #maxLineBytes: number; + readonly #maxPendingRequests: number; + readonly #maxInflightServerRequests: number; + readonly #maxBufferedOutputBytes: number; + readonly #stdoutDecoder: BoundedLineDecoder; + readonly #stderrDecoder: BoundedLineDecoder; + + constructor(options: ProcessCodexTransportOptions = {}) { + this.#onDiagnostic = options.onDiagnostic; + this.#maxLineBytes = positiveLimit( + options.maxLineBytes, + DEFAULT_MAX_LINE_BYTES, + ); + this.#maxPendingRequests = positiveLimit( + options.maxPendingRequests, + DEFAULT_MAX_PENDING_REQUESTS, + ); + this.#maxInflightServerRequests = positiveLimit( + options.maxInflightServerRequests, + DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS, + ); + this.#maxBufferedOutputBytes = positiveLimit( + options.maxBufferedOutputBytes, + DEFAULT_MAX_BUFFERED_OUTPUT_BYTES, + ); + this.#processGroup = + options.processGroup === true && process.platform !== "win32"; + this.#startedAt = new Date().toISOString(); + this.#closeGraceMs = positiveLimit(options.closeGraceMs, 1_000); + this.#onProcess = options.onProcess; + this.#exitPromise = new Promise((resolve) => { + this.#resolveExit = resolve; + }); + this.#notifications = new BoundedAsyncQueue( + positiveLimit( + options.maxQueuedNotifications, + DEFAULT_MAX_QUEUED_NOTIFICATIONS, + ), + positiveLimit( + options.maxQueuedNotificationBytes, + DEFAULT_MAX_QUEUED_NOTIFICATION_BYTES, + ), + ); + this.#process = spawn( + options.command ?? "codex", + options.args ?? ["app-server"], + { + env: options.environment ?? createSanitizedCodexEnvironment(), + stdio: "pipe", + detached: this.#processGroup, + }, + ); + this.#stdoutDecoder = new BoundedLineDecoder( + this.#maxLineBytes, + (line, bytes) => this.#onLine(line, bytes), + (error) => this.#fatal(error), + ); + this.#stderrDecoder = new BoundedLineDecoder( + MAX_DIAGNOSTIC_LINE_BYTES, + (line) => this.#onDiagnostic?.(redactCodexDiagnostic(line)), + () => + this.#onDiagnostic?.( + "codex app-server diagnostic line exceeded retention limit", + ), + ); + this.#process.stdout.on("data", (chunk: Buffer) => + this.#stdoutDecoder.write(chunk), + ); + this.#process.stdout.on("end", () => { + this.#stdoutDecoder.end(); + this.#fatal(new Error("codex app-server stdout ended before transport closure")); + }); + this.#process.stdout.on("error", (error) => this.#fatal(error)); + this.#process.stdout.on("close", () => { + this.#fatal( + new Error("codex app-server stdout closed before transport closure"), + ); + }); + this.#process.stderr.on("data", (chunk: Buffer) => + this.#stderrDecoder.write(chunk), + ); + this.#process.stderr.on("end", () => this.#stderrDecoder.end()); + this.#process.stdin.on("error", (error) => this.#fatal(error)); + this.#process.on("error", (error) => this.#fatal(error)); + this.#process.on("exit", (code, signal) => { + this.#exited = true; + this.#exitCode = code; + this.#exitSignal = signal; + this.#resolveExit(); + this.#onProcess?.(this.processInfo()); + if (!this.#closed) { + this.#fatal( + new Error( + `codex app-server exited (code=${code ?? "null"}, signal=${signal ?? "none"})`, + ), + ); + } + }); + this.#onProcess?.(this.processInfo()); + } + + request( + method: string, + params: Record, + ): Promise> { + if (this.#closed) + return Promise.reject(new Error("codex app-server transport is closed")); + if (this.#pending.size >= this.#maxPendingRequests) { + return Promise.reject( + new Error( + `codex app-server pending request limit ${this.#maxPendingRequests} exceeded`, + ), + ); + } + const id = this.#nextId++; + return new Promise((resolve, reject) => { + this.#pending.set(id, { resolve, reject }); + try { + this.#write({ method, id, params }); + } catch (error) { + this.#pending.delete(id); + reject(error); + } + }); + } + + notify(method: string, params?: Record): void { + this.#write(params === undefined ? { method } : { method, params }); + } + + notifications(): AsyncIterable { + return this.#notifications; + } + + setServerRequestHandler(handler: CodexServerRequestHandler): void { + this.#serverRequestHandler = handler; + } + + async close(): Promise { + const firstClose = !this.#closed; + if (firstClose) { + this.#closed = true; + this.#stdoutDecoder.close(); + this.#stderrDecoder.close(); + this.#notifications.close(); + this.#failAll(new Error("codex app-server transport closed")); + } + if (!this.#exited) { + if (firstClose) this.#signal("SIGTERM"); + const exited = await Promise.race([ + this.#exitPromise.then(() => true), + new Promise((resolve) => + setTimeout(() => resolve(false), this.#closeGraceMs), + ), + ]); + if (!exited && !this.#exited) { + this.#signal("SIGKILL"); + await Promise.race([ + this.#exitPromise, + new Promise((resolve) => + setTimeout(resolve, this.#closeGraceMs), + ), + ]); + } + } + } + + processInfo(): CodexTransportProcessInfo { + return { + pid: this.#process.pid ?? null, + processGroupId: this.#processGroup ? (this.#process.pid ?? null) : null, + startedAt: this.#startedAt, + exited: this.#exited, + exitCode: this.#exitCode, + signal: this.#exitSignal, + }; + } + + #signal(signal: NodeJS.Signals): void { + const pid = this.#process.pid; + if (this.#processGroup && pid !== undefined) { + try { + process.kill(-pid, signal); + return; + } catch { + // The group may already be gone; the direct child kill below is idempotent. + } + } + this.#process.kill(signal); + } + + #write(message: unknown): void { + if (this.#closed) throw new Error("codex app-server transport is closed"); + let serialized: string; + try { + serialized = JSON.stringify(message); + } catch (error) { + const failure = error instanceof Error ? error : new Error(String(error)); + this.#fatal(failure); + throw failure; + } + const line = `${serialized}\n`; + const lineBytes = Buffer.byteLength(line); + if (lineBytes - 1 > this.#maxLineBytes) { + const error = new Error( + `outbound codex JSON-RPC line exceeded ${this.#maxLineBytes} bytes`, + ); + this.#fatal(error); + throw error; + } + if ( + this.#process.stdin.writableLength + lineBytes > + this.#maxBufferedOutputBytes + ) { + const error = new Error( + `outbound codex JSON-RPC buffer exceeded ${this.#maxBufferedOutputBytes} bytes`, + ); + this.#fatal(error); + throw error; + } + try { + this.#process.stdin.write(line); + } catch (error) { + const failure = error instanceof Error ? error : new Error(String(error)); + this.#fatal(failure); + throw failure; + } + } + + #onLine(line: string, bytes: number): void { + let value: unknown; + try { + value = JSON.parse(line); + } catch { + this.#fatal(new Error("codex app-server emitted malformed JSON")); + return; + } + if (!isRecord(value)) { + this.#fatal( + new Error("codex app-server emitted a non-object JSON-RPC message"), + ); + return; + } + const message = value; + const hasResult = Object.hasOwn(message, "result"); + const hasError = Object.hasOwn(message, "error"); + if (typeof message.id === "number" && (hasResult || hasError)) { + if ( + !Number.isSafeInteger(message.id) || + message.id <= 0 || + hasResult === hasError + ) { + this.#fatal( + new Error("codex app-server emitted a malformed JSON-RPC response"), + ); + return; + } + const pending = this.#pending.get(message.id); + if (pending === undefined) { + this.#fatal( + new Error("codex app-server responded to an unknown request id"), + ); + return; + } + this.#pending.delete(message.id); + if (hasError) { + pending.reject( + new CodexRpcError( + redactCodexDiagnostic(boundedJson(message.error)), + rpcErrorCode(message.error), + ), + ); + } else if (isRecord(message.result)) { + pending.resolve(message.result); + } else { + const error = new Error( + "codex app-server response result was not an object", + ); + pending.reject(error); + this.#fatal(error); + } + return; + } + if ( + typeof message.method !== "string" || + message.method.length === 0 || + hasResult || + hasError + ) { + this.#fatal( + new Error("codex app-server emitted a malformed JSON-RPC message"), + ); + return; + } + if (message.params !== undefined && !isRecord(message.params)) { + this.#fatal( + new Error("codex app-server message params were not an object"), + ); + return; + } + const params = message.params ?? {}; + if (message.id !== undefined) { + if (!isValidServerRequestId(message.id)) { + this.#fatal( + new Error("codex app-server request id had an invalid type"), + ); + return; + } + this.#handleServerRequest({ + id: message.id, + method: message.method, + params, + }); + return; + } + if (!this.#notifications.push({ method: message.method, params }, bytes)) { + this.#fatal( + new Error("codex app-server notification queue limit exceeded"), + ); + } + } + + #handleServerRequest(request: CodexRpcServerRequest): void { + if (this.#inflightServerRequests >= this.#maxInflightServerRequests) { + this.#writeServerResponse({ + id: request.id, + error: { + code: -32_001, + message: "Client request queue limit exceeded.", + }, + }); + return; + } + this.#inflightServerRequests += 1; + void Promise.resolve() + .then(() => this.#serverRequestHandler(request)) + .then( + (result) => { + this.#writeServerResponse({ id: request.id, result }); + }, + (error: unknown) => { + this.#writeServerResponse({ + id: request.id, + error: { + code: -32_000, + message: redactCodexDiagnostic(String(error)), + }, + }); + }, + ) + .finally(() => { + this.#inflightServerRequests -= 1; + }); + } + + #writeServerResponse(message: unknown): void { + if (this.#closed) return; + try { + this.#write(message); + } catch (error) { + this.#fatal(error instanceof Error ? error : new Error(String(error))); + } + } + + #fatal(error: Error): void { + if (this.#closed) return; + this.#closed = true; + this.#stdoutDecoder.close(); + this.#stderrDecoder.close(); + this.#signal("SIGKILL"); + this.#notifications.close(error); + this.#failAll(error); + this.#onDiagnostic?.(redactCodexDiagnostic(error.message)); + } + + #failAll(error: Error): void { + for (const pending of this.#pending.values()) pending.reject(error); + this.#pending.clear(); + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function positiveLimit(value: number | undefined, fallback: number): number { + return Number.isSafeInteger(value) && (value ?? 0) > 0 ? value! : fallback; +} + +function isValidServerRequestId(value: unknown): value is string | number { + return ( + (typeof value === "number" && Number.isSafeInteger(value)) || + (typeof value === "string" && + value.length > 0 && + Buffer.byteLength(value) <= 256) + ); +} + +function boundedJson(value: unknown): string { + const serialized = JSON.stringify(value) ?? String(value); + return serialized.length <= MAX_DIAGNOSTIC_LINE_BYTES + ? serialized + : `${serialized.slice(0, MAX_DIAGNOSTIC_LINE_BYTES)}...[truncated]`; +}