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 cc32a3c013..3018461ba9 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 @@ -183,3 +183,29 @@ describe("Codex app-server transport", () => { codex.child.stdin.end(); }); }); + +describe("Codex stdout byte admission", () => { + test("fails pending requests and the session before joining an oversized line", async () => { + const child = createCodexAppServerChildProcess(); + const client = new CodexAppServerClient(child, createTestLogger()); + const terminations: Error[] = []; + client.setUnexpectedTerminationHandler((error) => terminations.push(error)); + const request = client.request("thread/resume", {}, 1000).catch((error: Error) => error); + const chunk = Buffer.alloc(64 * 1024, "x"); + child.stdout.write('{"id":1,"result":"'); + for (let index = 0; index < 2048; index++) child.stdout.write(chunk); + const error = await request; + expect(error).toMatchObject({ + name: "ProviderStreamLimitError", + code: "PROVIDER_STREAM_LINE_TOO_LARGE", + limitBytes: 128 * 1024 * 1024, + }); + expect(terminations).toEqual([error]); + expect(client.hasPendingRequests()).toBe(false); + child.stdout.write('"}\n{"id":1,"result":"late"}\n'); + await expect(client.request("model/list")).rejects.toThrow("closed"); + await client.dispose(); + child.stdout.end(); + child.stderr.end(); + }); +}); 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 29b8cf8880..dfd4adfac4 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,10 +1,15 @@ import type { ChildProcessWithoutNullStreams } from "node:child_process"; -import { StringDecoder } from "node:string_decoder"; import type { Logger } from "pino"; import { z } from "zod"; import { terminateWithTreeKill } from "../../../../utils/tree-kill.js"; +import { ProviderLineReader, ProviderStreamLimitError } from "../jsonl-frame-decoder.js"; + +// A measured long-thread resume was 19.75 MB; preserve the existing 32 MiB fixture +// with 4x margin for longer threads. This is provider stdout, not the client protocol. +export const CODEX_PROVIDER_LINE_BYTES = 128 * 1024 * 1024; + const DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000; const APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 2_000; const APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; @@ -167,39 +172,6 @@ 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 stdoutLineReader: { close(): void }; private activeProviderRequests = 0; @@ -216,11 +188,24 @@ export class CodexAppServerClient { private readonly logger: Logger, private readonly getTraceContext: () => CodexAppServerTraceContext = () => ({}), ) { - 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"); - }); - }); + // Split only on LF: U+2028/U+2029 are valid raw characters inside Codex JSON. + const reader = new ProviderLineReader( + CODEX_PROVIDER_LINE_BYTES, + (line) => { + void this.handleLine(line).catch((error) => { + this.logger.warn({ error, line }, "Failed to handle Codex app-server stdout line"); + }); + }, + (error) => this.handleStreamLimit(error), + ); + const onData = (chunk: Buffer | string) => reader.write(chunk); + child.stdout.on("data", onData); + this.stdoutLineReader = { + close: () => { + reader.close(); + child.stdout.off("data", onData); + }, + }; child.stderr.on("data", (chunk) => { this.stderrBuffer += chunk.toString(); @@ -318,6 +303,15 @@ export class CodexAppServerClient { } } + private handleStreamLimit(failure: ProviderStreamLimitError): void { + // A rejected response/notification cannot be replayed safely. Fail the session, + // including live turns with no pending RPC, rather than dropping a stream unit. + this.handleUnexpectedTermination(failure); + void this.dispose().catch((err) => { + this.logger.warn({ err }, "Failed to terminate oversized Codex stdout producer"); + }); + } + private handleUnexpectedTermination(error: Error): void { if (this.disposed) { return; diff --git a/packages/server/src/server/agent/providers/jsonl-frame-decoder.test.ts b/packages/server/src/server/agent/providers/jsonl-frame-decoder.test.ts index af2bdabf01..c1c81bdae8 100644 --- a/packages/server/src/server/agent/providers/jsonl-frame-decoder.test.ts +++ b/packages/server/src/server/agent/providers/jsonl-frame-decoder.test.ts @@ -1,3 +1,5 @@ +import { performance } from "node:perf_hooks"; +import { setTimeout as delay } from "node:timers/promises"; import { describe, expect, it } from "vitest"; import { JsonlFrameDecoder, type JsonlFrameProblem } from "./jsonl-frame-decoder.js"; @@ -36,6 +38,16 @@ describe("JsonlFrameDecoder", () => { expect(problems).toEqual([]); }); + it("preserves UTF-8 characters split across raw stdout buffers", () => { + const { decoder, frames, problems } = decoderHarness(); + const bytes = Buffer.from('{"type":"notice","text":"é🙂"}\n'); + const split = bytes.indexOf(Buffer.from("🙂")) + 1; + decoder.write(bytes.subarray(0, split)); + decoder.write(bytes.subarray(split)); + expect(frames).toEqual([{ type: "notice", text: "é🙂" }]); + expect(problems).toEqual([]); + }); + it("reassembles a large ordered v2 frame", () => { const { decoder, frames, problems } = decoderHarness(); const frame = { type: "models", value: "x".repeat(1024 * 1024) }; @@ -55,3 +67,57 @@ describe("JsonlFrameDecoder", () => { expect(problems).toEqual(["out-of-order-chunk"]); }); }); + +describe("provider line admission", () => { + it("keeps the loop responsive while accumulating a fragmented provider response", async () => { + const { decoder, frames, problems } = decoderHarness(); + const part = "x".repeat(512); + let last = performance.now(); + let longestGap = 0; + const probe = setInterval(() => { + const now = performance.now(); + longestGap = Math.max(longestGap, now - last); + last = now; + }, 1); + try { + await delay(5); + decoder.write('{"type":"notice","text":"'); + // 1 MiB is enough to expose the former rescan of every accumulated prefix. + for (let index = 0; index < 2048; index++) decoder.write(part); + decoder.write('"}\n'); + await delay(5); + expect(frames).toEqual([{ type: "notice", text: part.repeat(2048) }]); + expect(problems).toEqual([]); + expect(longestGap).toBeLessThan(50); + } finally { + clearInterval(probe); + } + }); + + it("rejects an unterminated over-cap UTF-8 line and stops accepting frames", () => { + const frames: Record[] = []; + const failures: Array<{ problem: string; detail: unknown }> = []; + const decoder = new JsonlFrameDecoder({ + frame: (frame) => frames.push(frame), + problem: (problem, detail) => failures.push({ problem, detail }), + }); + // 64 MiB of UTF-8, only 32 Mi UTF-16 code units. No newline or JSON parse needed. + const chunk = "é".repeat(32 * 1024); + for (let index = 0; index < 1024; index++) decoder.write(chunk); + expect(failures).toEqual([]); + decoder.write("é"); + expect(failures).toEqual([ + { + problem: "line-too-large", + detail: expect.objectContaining({ + name: "ProviderStreamLimitError", + code: "PROVIDER_STREAM_LINE_TOO_LARGE", + limitBytes: 64 * 1024 * 1024, + }), + }, + ]); + decoder.write('\n{"type":"ready"}\n'); + expect(frames).toEqual([]); + expect(failures).toHaveLength(1); + }); +}); diff --git a/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts b/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts index 1662e7719c..8b1b9767de 100644 --- a/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts +++ b/packages/server/src/server/agent/providers/jsonl-frame-decoder.ts @@ -1,3 +1,60 @@ +// Provider-process stdout only; these are not app/daemon wire limits. +export const JSONL_PROVIDER_LINE_BYTES = 64 * 1024 * 1024; + +export class ProviderStreamLimitError extends Error { + readonly code = "PROVIDER_STREAM_LINE_TOO_LARGE"; + + constructor(readonly limitBytes: number) { + super( + `Provider stdout line exceeds ${limitBytes} bytes. Start a new thread or reduce the provider response size.`, + ); + this.name = "ProviderStreamLimitError"; + } +} + +/** Admits raw UTF-8 bytes before retaining, decoding or joining a complete line. */ +export class ProviderLineReader { + private parts: Buffer[] = []; + private bytes = 0; + private stopped = false; + + constructor( + private readonly limitBytes: number, + private readonly onLine: (line: string) => void, + private readonly onLimit: (error: ProviderStreamLimitError) => void, + ) {} + + write(chunk: Buffer | string): void { + if (this.stopped) return; + const buffer = typeof chunk === "string" ? Buffer.from(chunk, "utf8") : chunk; + let start = 0; + while (start < buffer.byteLength && !this.stopped) { + const newline = buffer.indexOf(10, start); + const end = newline === -1 ? buffer.byteLength : newline; + const length = end - start; + if (length > this.limitBytes - this.bytes) { + this.close(); + this.onLimit(new ProviderStreamLimitError(this.limitBytes)); + return; + } + if (length) this.parts.push(buffer.subarray(start, end)); + this.bytes += length; + if (newline === -1) return; + const line = Buffer.concat(this.parts, this.bytes).toString("utf8"); + this.parts = []; + this.bytes = 0; + this.onLine(line.endsWith("\r") ? line.slice(0, -1) : line); + start = newline + 1; + } + } + + close(): void { + this.stopped = true; + this.parts = []; + this.bytes = 0; + } +} + const V2_FRAME_BYTES = 1024 * 1024; const V2_REASSEMBLED_BYTES = 64 * 1024 * 1024; const V2_CHUNK_BYTES = 256 * 1024; @@ -5,6 +62,7 @@ const V2_BASE64_CHARS = Math.ceil(V2_CHUNK_BYTES / 3) * 4; const V2_MAX_CHUNKS = Math.ceil(V2_REASSEMBLED_BYTES / V2_CHUNK_BYTES); export type JsonlFrameProblem = + | "line-too-large" | "invalid-json" | "invalid-chunk" | "invalid-base64" @@ -78,7 +136,7 @@ function chunkBytes(data: string): Buffer | null { } export class JsonlFrameDecoder { - private lineBuffer = ""; + private readonly lines: ProviderLineReader; private assembly: ChunkAssembly | null = null; constructor( @@ -86,17 +144,18 @@ export class JsonlFrameDecoder { frame(message: Record): void; problem(problem: JsonlFrameProblem, detail?: unknown): void; }, - ) {} + ) { + this.lines = new ProviderLineReader( + JSONL_PROVIDER_LINE_BYTES, + (line) => { + if (line.trim()) this.consumeLine(line); + }, + (error) => this.drop("line-too-large", error), + ); + } - write(text: string): void { - this.lineBuffer += text; - let newline = this.lineBuffer.indexOf("\n"); - while (newline !== -1) { - const line = this.lineBuffer.slice(0, newline).replace(/\r$/, ""); - this.lineBuffer = this.lineBuffer.slice(newline + 1); - if (line.trim()) this.consumeLine(line); - newline = this.lineBuffer.indexOf("\n"); - } + write(chunk: Buffer | string): void { + this.lines.write(chunk); } private consumeLine(line: string): void { diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts index 7dfbb6abb8..141b8f36f0 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts @@ -471,3 +471,25 @@ describe("JsonlRpcProcess", () => { } }); }); + +describe("JSONL stdout byte admission", () => { + test("fails pending requests and publishes the typed failure to live-turn consumers", async () => { + const child = createInMemoryChildProcess(); + const process = startProcess({ child }); + const exit = new Promise((resolve) => process.onExit(resolve)); + const request = process.request({ type: "hang" }, 1000).catch((error: Error) => error); + const chunk = Buffer.alloc(64 * 1024, "x"); + child.stdout.write('{"type":"notice","text":"'); + for (let index = 0; index < 1024; index++) child.stdout.write(chunk); + const error = await request; + expect(error).toMatchObject({ + name: "ProviderStreamLimitError", + code: "PROVIDER_STREAM_LINE_TOO_LARGE", + limitBytes: 64 * 1024 * 1024, + }); + expect((await exit).error).toBe(error); + await process.close(); + child.stdout.end(); + child.stderr.end(); + }); +}); diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts index 9d8b9af623..ddf85eb316 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts @@ -3,7 +3,7 @@ import type { Logger } from "pino"; import { spawnProcess } from "../../../utils/spawn.js"; import { terminateWithTreeKill } from "../../../utils/tree-kill.js"; -import { JsonlFrameDecoder } from "./jsonl-frame-decoder.js"; +import { JsonlFrameDecoder, ProviderStreamLimitError } from "./jsonl-frame-decoder.js"; export { supportsJsonlRpcProtocolV2 } from "./jsonl-frame-decoder.js"; /** Default wall-clock timeout for control-plane / short RPC calls. */ @@ -81,6 +81,7 @@ export class JsonlRpcProcess { private stderrBuffer = ""; private nextRequestId = 1; private disposed = false; + private streamFailure: ProviderStreamLimitError | null = null; private exited = false; private closing: Promise | null = null; private readonly frameDecoder: JsonlFrameDecoder; @@ -90,6 +91,14 @@ export class JsonlRpcProcess { this.frameDecoder = new JsonlFrameDecoder({ frame: (message) => this.dispatchFrame(message), problem: (problem, detail) => { + if (problem === "line-too-large" && detail instanceof ProviderStreamLimitError) { + // Stop the producer: silently skipping a response or turn event loses state. + this.streamFailure = detail; + void this.close(detail).catch((err) => { + this.options.logger.warn({ err }, `Failed to close ${this.diagnosticName}`); + }); + return; + } this.options.logger.warn( { problem, detail }, `Ignoring invalid ${this.diagnosticName} frame`, @@ -98,7 +107,7 @@ export class JsonlRpcProcess { }); this.child = (options.spawn ?? spawnJsonlRpcProcess)(options.launch); this.child.stdout.on("data", (chunk) => { - this.handleStdoutChunk(chunk.toString()); + this.handleStdoutChunk(chunk); }); this.child.stderr.on("data", (chunk) => { this.stderrBuffer += chunk.toString(); @@ -114,9 +123,11 @@ export class JsonlRpcProcess { }); this.child.on("exit", (code, signal) => { this.exited = true; - const error = new Error( - `${this.diagnosticName} process exited with code ${code ?? "null"} and signal ${signal ?? "null"}\n${this.stderrBuffer}`.trim(), - ); + const error = + this.streamFailure ?? + new Error( + `${this.diagnosticName} process exited with code ${code ?? "null"} and signal ${signal ?? "null"}\n${this.stderrBuffer}`.trim(), + ); const exit = { code, signal, error }; for (const subscriber of this.exitSubscribers) { subscriber(exit); @@ -254,7 +265,7 @@ export class JsonlRpcProcess { } } - private handleStdoutChunk(chunk: string): void { + private handleStdoutChunk(chunk: Buffer | string): void { this.frameDecoder.write(chunk); } diff --git a/packages/server/src/server/workspace-registry.test.ts b/packages/server/src/server/workspace-registry.test.ts index 6c27027edc..e288629656 100644 --- a/packages/server/src/server/workspace-registry.test.ts +++ b/packages/server/src/server/workspace-registry.test.ts @@ -116,6 +116,35 @@ describe("workspace registries", () => { expect(await projectRegistry.list()).toEqual([]); }); + test("writes once per mutation and skips missing-record mutations", async () => { + let writes = 0; + const registry = new FileBackedProjectRegistry( + path.join(tmpDir, "projects", "write-count.json"), + logger, + { + writeRecords: async (filePath, records) => { + writes++; + await writeJsonFileAtomic(filePath, records); + }, + }, + ); + const project = createPersistedProjectRecord({ + projectId: "counted", + rootPath: tmpDir, + kind: "git", + displayName: "Counted", + createdAt: "2026-03-01T00:00:00.000Z", + updatedAt: "2026-03-01T00:00:00.000Z", + }); + await registry.upsert(project); + expect(writes).toBe(1); + await registry.update("missing", (current) => current); + await registry.remove("missing"); + expect(writes).toBe(1); + await registry.update(project.projectId, (current) => ({ ...current, customName: "Changed" })); + expect(writes).toBe(2); + }); + test("preserves a concurrent project update when archiving", async () => { let pauseNextWrite = false; let releaseWrite!: () => void;