fix(server): reject oversized provider stdout before parsing

Admit raw UTF-8 line bytes before accumulation and fail the provider session explicitly on overflow. Share line framing so fragmented JSONL input no longer repeatedly scans and copies the complete prefix.

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();
+  });
+});

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;

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<string, unknown>[] = [];
+    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);
+  });
+});

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<string, unknown>): 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 {

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<JsonlRpcExit>((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();
+  });
+});

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<void> | 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);
   }
 

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;