diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts index 5c9dbe956c..cc32a3c013 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts @@ -22,6 +22,44 @@ describe("Codex app-server transport", () => { child.stdin.end(); }); + test("keeps a JSON message intact when it contains raw U+2028/U+2029", async () => { + const child = createCodexAppServerChildProcess(); + const client = new CodexAppServerClient(child, createTestLogger()); + + const request = client.request("thread/resume", {}); + const text = "a\u2028b\u2029c\u0085d"; + // Codex does not escape U+2028/U+2029 in JSON output, so write them raw. + const message = `{"id":1,"result":{"text":"${text}"}}\n`; + const bytes = Buffer.from(message, "utf8"); + // Split mid-way (inside a multi-byte character) to exercise chunk boundaries. + child.stdout.write(bytes.subarray(0, 28)); + child.stdout.write(bytes.subarray(28)); + + await expect(request).resolves.toEqual({ text }); + child.stdout.end(); + child.stderr.end(); + child.stdin.end(); + }); + + test("reads a multi-megabyte JSON message arriving in 64 KB chunks", async () => { + const child = createCodexAppServerChildProcess(); + const client = new CodexAppServerClient(child, createTestLogger()); + + const request = client.request("thread/resume", {}); + // A resumed long thread arrives as one large line, split into 64 KB stdout chunks. + const text = "x".repeat(32 * 1024 * 1024); + const bytes = Buffer.from(`{"id":1,"result":{"text":"${text}"}}\n`, "utf8"); + const chunkSize = 64 * 1024; + for (let offset = 0; offset < bytes.length; offset += chunkSize) { + child.stdout.write(bytes.subarray(offset, offset + chunkSize)); + } + + await expect(request).resolves.toEqual({ text }); + child.stdout.end(); + child.stderr.end(); + child.stdin.end(); + }); + test("dispose rejects pending requests instead of leaving them hanging", async () => { const child = createCodexAppServerChildProcess(); const client = new CodexAppServerClient(child, createTestLogger()); diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.ts index 1720118c63..acb5e76341 100644 --- a/packages/server/src/server/agent/providers/codex/app-server-transport.ts +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.ts @@ -1,5 +1,5 @@ import type { ChildProcessWithoutNullStreams } from "node:child_process"; -import readline from "node:readline"; +import { StringDecoder } from "node:string_decoder"; import type { Logger } from "pino"; import { z } from "zod"; @@ -167,8 +167,41 @@ function readProviderTurnId(params: unknown): string | undefined { return isRecord(turn) && typeof turn.id === "string" ? turn.id : undefined; } +/** + * Splits stdout into lines on "\n" only. `node:readline` additionally treats U+2028 and U+2029 + * as line terminators, but Codex emits those characters raw inside JSON strings, so readline + * would cut a single JSON-RPC message into pieces that fail to parse. + */ +function createNewlineLineReader( + input: NodeJS.ReadableStream, + onLine: (line: string) => void, +): { close(): void } { + const decoder = new StringDecoder("utf8"); + // Text after the last "\n", kept as parts and joined once its line ends, so a large line + // arriving in many chunks is scanned and copied once instead of once per chunk. + let unterminated: string[] = []; + const onData = (chunk: Buffer | string): void => { + const text = typeof chunk === "string" ? chunk : decoder.write(chunk); + let lineStart = 0; + let index = text.indexOf("\n"); + while (index !== -1) { + unterminated.push(text.slice(lineStart, index)); + const line = unterminated.join(""); + unterminated = []; + onLine(line.endsWith("\r") ? line.slice(0, -1) : line); + lineStart = index + 1; + index = text.indexOf("\n", lineStart); + } + if (lineStart < text.length) { + unterminated.push(text.slice(lineStart)); + } + }; + input.on("data", onData); + return { close: () => void input.off("data", onData) }; +} + export class CodexAppServerClient { - private readonly rl: readline.Interface; + private readonly stdoutLineReader: { close(): void }; private readonly pending = new Map(); private readonly requestHandlers = new Map(); private notificationHandler: NotificationHandler | null = null; @@ -182,8 +215,7 @@ export class CodexAppServerClient { private readonly logger: Logger, private readonly getTraceContext: () => CodexAppServerTraceContext = () => ({}), ) { - this.rl = readline.createInterface({ input: child.stdout }); - this.rl.on("line", (line) => { + this.stdoutLineReader = createNewlineLineReader(child.stdout, (line) => { void this.handleLine(line).catch((error) => { this.logger.warn({ error, line }, "Failed to handle Codex app-server stdout line"); }); @@ -259,7 +291,7 @@ export class CodexAppServerClient { async dispose(): Promise { this.disposed = true; this.unexpectedTerminationHandler = null; - this.rl.close(); + this.stdoutLineReader.close(); this.rejectPending(new Error("Codex app-server client is closed")); try { this.child.stdin.end(); @@ -286,7 +318,7 @@ export class CodexAppServerClient { return; } this.disposed = true; - this.rl.close(); + this.stdoutLineReader.close(); this.rejectPending(error); const handler = this.unexpectedTerminationHandler; this.unexpectedTerminationHandler = null;