From 3dafe91d24f5945969af5489b7123860de3f29db Mon Sep 17 00:00:00 2001 From: Victor Vogel <263261067+only21mil@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:16:12 -0500 Subject: [PATCH 1/5] fix(codex): oversized app-server messages end the session before they are parsed The Codex app-server reader kept every fragment of an unfinished JSONL line and joined and parsed it with no size check, so one huge provider message could exhaust the heap. The reader now counts each message's UTF-8 bytes as fragments arrive and, past a 128 MiB ceiling (excluding the line terminator), drops the buffered fragments and ends the input stream with a read-input-stream transport error that carries only sizes. Complete lines that arrived earlier in the same chunk are still dispatched first. Co-authored-by: Cestercian <183791452+cestercian@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- .../src/protocol.test.ts | 235 ++++++++++++++++++ .../effect-codex-app-server/src/protocol.ts | 87 +++++-- 2 files changed, 307 insertions(+), 15 deletions(-) diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 7249afff10713..249924aa1f32c 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -34,6 +34,49 @@ const decodeConsumeRateLimitResetCreditResponse = Schema.decodeUnknownEffect( CodexRpc.CLIENT_REQUEST_RESPONSES["account/rateLimitResetCredit/consume"], ); +const lineFramings = [ + { label: "LF", frame: (line: string) => [`${line}\n`] }, + { label: "CRLF", frame: (line: string) => [`${line}\r\n`] }, + { label: "CRLF split across chunks", frame: (line: string) => [`${line}\r`, "\n"] }, + { label: "CRLF split by an empty chunk", frame: (line: string) => [`${line}\r`, "", "\n"] }, +]; + +const makeLimitedProtocol = (maxIncomingMessageBytes: number) => + Effect.gen(function* () { + const { stdio, input, output } = yield* makeInMemoryStdio(); + const termination = yield* Deferred.make(); + const terminations: Array = []; + const notifications: Array = []; + const events: Array = []; + const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + maxIncomingMessageBytes, + logIncoming: true, + logger: (event) => + Effect.sync(() => { + events.push(event); + }), + onNotification: (notification) => + Effect.sync(() => { + notifications.push(notification); + }), + onTermination: (error) => + Effect.sync(() => terminations.push(error)).pipe( + Effect.andThen(Deferred.succeed(termination, error)), + Effect.asVoid, + ), + }); + return { input, output, transport, termination, terminations, notifications, events }; + }); + +const awaitMessageLimitFailure = (termination: Deferred.Deferred) => + Effect.gen(function* () { + const error = yield* Deferred.await(termination); + assert.instanceOf(error, CodexError.CodexAppServerTransportError); + assert.equal(error.operation, "read-input-stream"); + return error; + }); + it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { it.effect("maps account usage responses to the upstream token usage schema", () => Effect.gen(function* () { @@ -325,6 +368,198 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); + it.effect.each(lineFramings)( + "accepts a message exactly at the byte limit with $label framing", + ({ frame }) => + Effect.gen(function* () { + const message = { method: "x/limit", params: { text: "at the limit" } }; + const line = encodeUnknownJsonString(message); + const { input, termination, notifications } = yield* makeLimitedProtocol( + encoder.encode(line).length, + ); + + for (const chunk of frame(line)) { + yield* Queue.offer(input, encoder.encode(chunk)); + } + yield* Queue.end(input); + + assert.instanceOf( + yield* Deferred.await(termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(notifications, [message]); + }), + ); + + it.effect.each(lineFramings)( + "rejects a message one byte over the limit with $label framing", + ({ frame }) => + Effect.gen(function* () { + const line = encodeUnknownJsonString({ method: "x/over", params: { text: "one over" } }); + const { input, output, transport, termination, terminations, notifications, events } = + yield* makeLimitedProtocol(encoder.encode(line).length - 1); + const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped); + yield* Queue.take(output); + + for (const chunk of frame(line)) { + yield* Queue.offer(input, encoder.encode(chunk)); + } + yield* Queue.end(input); + + const error = yield* awaitMessageLimitFailure(termination); + const pendingError = yield* Fiber.join(pending).pipe( + Effect.match({ + onFailure: (failure) => failure, + onSuccess: () => assert.fail("Expected the oversized message to fail the request"), + }), + ); + assert.strictEqual(pendingError, error); + assert.lengthOf(terminations, 1); + assert.strictEqual(terminations[0], error); + assert.deepEqual(notifications, []); + assert.deepEqual(events, []); + }), + ); + + it.effect("rejects an over-limit message from small fragments before its newline arrives", () => + Effect.gen(function* () { + const { input, output, transport, termination, notifications, events } = + yield* makeLimitedProtocol(1024); + const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped); + yield* Queue.take(output); + + yield* Queue.offer(input, encoder.encode('{"method":"x/huge","params":{"text":"')); + for (let index = 0; index < 512; index++) { + yield* Queue.offer(input, encoder.encode("xxxxxxxx")); + } + + const error = yield* awaitMessageLimitFailure(termination); + const pendingError = yield* Fiber.join(pending).pipe( + Effect.match({ + onFailure: (failure) => failure, + onSuccess: () => assert.fail("Expected the oversized message to fail the request"), + }), + ); + assert.strictEqual(pendingError, error); + assert.deepEqual(notifications, []); + assert.deepEqual(events, []); + }), + ); + + it.effect("routes an unterminated final message at the byte limit when input ends", () => + Effect.gen(function* () { + const message = { method: "x/final", params: { text: "no newline" } }; + const line = encodeUnknownJsonString(message); + const { input, termination, notifications } = yield* makeLimitedProtocol( + encoder.encode(line).length, + ); + + yield* Queue.offer(input, encoder.encode(line)); + yield* Queue.end(input); + + assert.instanceOf( + yield* Deferred.await(termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(notifications, [message]); + }), + ); + + it.effect.each([ + { label: "an unterminated final message", suffix: "" }, + { label: "a final message ending in a bare \\r", suffix: "\r" }, + ])("rejects $label one byte over the limit when input ends", ({ suffix }) => + Effect.gen(function* () { + const line = encodeUnknownJsonString({ method: "x/final", params: { text: "no newline" } }); + const { input, termination, notifications, events } = yield* makeLimitedProtocol( + encoder.encode(line).length + suffix.length - 1, + ); + + yield* Queue.offer(input, encoder.encode(`${line}${suffix}`)); + yield* Queue.end(input); + + yield* awaitMessageLimitFailure(termination); + assert.deepEqual(notifications, []); + assert.deepEqual(events, []); + }), + ); + + it.effect.each([1, 7])( + "counts UTF-8 bytes of characters split across %i-byte input chunks", + (chunkSize) => + Effect.gen(function* () { + const message = { method: "x/unicode", params: { text: "ζœ€εΎŒπŸ™‚πŸ™‚ζœ€εΎŒ" } }; + const line = encodeUnknownJsonString(message); + const bytes = encoder.encode(`${line}\n`); + const lineBytes = bytes.length - 1; + // Counting UTF-16 code units instead of bytes would accept the over-limit message. + assert.isBelow(line.length, lineBytes - 1); + + const atLimit = yield* makeLimitedProtocol(lineBytes); + const overLimit = yield* makeLimitedProtocol(lineBytes - 1); + for (let offset = 0; offset < bytes.length; offset += chunkSize) { + yield* Queue.offer(atLimit.input, bytes.subarray(offset, offset + chunkSize)); + yield* Queue.offer(overLimit.input, bytes.subarray(offset, offset + chunkSize)); + } + yield* Queue.end(atLimit.input); + yield* Queue.end(overLimit.input); + + assert.instanceOf( + yield* Deferred.await(atLimit.termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(atLimit.notifications, [message]); + yield* awaitMessageLimitFailure(overLimit.termination); + assert.deepEqual(overLimit.notifications, []); + assert.deepEqual(overLimit.events, []); + }), + ); + + it.effect("keeps an over-limit message out of protocol logs and the termination error", () => + Effect.gen(function* () { + const secret = "codex-oversized-secret-sentinel"; + const small = { method: "x/small", params: {} }; + const { input, termination, notifications, events } = yield* makeLimitedProtocol(64); + + yield* Queue.offer(input, encodeJsonl(small)); + yield* Queue.offer(input, encodeJsonl({ method: "x/huge", params: { text: secret } })); + yield* Queue.end(input); + + const error = yield* awaitMessageLimitFailure(termination); + assert.deepEqual(notifications, [small]); + assert.deepEqual( + events.map(({ stage }) => stage), + ["raw", "decoded"], + ); + assert.notInclude(encodeUnknownJsonString(events), secret); + assert.notInclude(error.message, secret); + assert.notInclude(String(error.cause), secret); + }), + ); + + it.effect.each([ + { label: "a terminated", newline: "\n" }, + { label: "an unterminated", newline: "" }, + ])("dispatches complete lines that precede $label over-limit line in one chunk", ({ newline }) => + Effect.gen(function* () { + const first = { method: "x/first", params: { index: 1 } }; + const second = { method: "x/second", params: { index: 2 } }; + const oversized = { method: "x/huge", params: { text: "y".repeat(64) } }; + const { input, termination, notifications } = yield* makeLimitedProtocol(64); + + yield* Queue.offer( + input, + encoder.encode( + `${[first, second, oversized].map((message) => encodeUnknownJsonString(message)).join("\n")}${newline}`, + ), + ); + yield* Queue.end(input); + + yield* awaitMessageLimitFailure(termination); + assert.deepEqual(notifications, [first, second]); + }), + ); + it.effect.each([1, 7, 1024])( "preserves JSONL framing and UTF-8 across %i-byte input chunks", (chunkSize) => diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 4a32973a988bd..97adb1511aee7 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -16,6 +16,24 @@ const isJsonRpcId = Schema.is(JsonRpcId); const isJsonRpcResponseEnvelope = Schema.is(JsonRpcResponseEnvelope); const isCodexAppServerError = Schema.is(CodexError.CodexAppServerError); const MAX_BUFFERED_RAW_MESSAGES = 32; +// Largest incoming JSONL message, in UTF-8 bytes without its line terminator. +// Codex diffs reach ~49M characters and grow when JSON-escaped; a line far +// past this ceiling would exhaust the heap during join and parse. +const MAX_INCOMING_MESSAGE_BYTES = 128 * 1024 * 1024; + +// UTF-8 size of `text.slice(start, end)` without encoding a copy. Text from a +// streaming TextDecoder holds no lone surrogates, so each surrogate is one half +// of a four-byte sequence. +const utf8ByteLength = (text: string, start: number, end: number) => { + let bytes = 0; + for (let index = start; index < end; index++) { + const code = text.charCodeAt(index); + if (code < 0x80) bytes += 1; + else if (code < 0x800 || (code >= 0xd800 && code <= 0xdfff)) bytes += 2; + else bytes += 3; + } + return bytes; +}; export interface CodexAppServerProtocolLogEvent { readonly direction: "incoming" | "outgoing"; @@ -39,6 +57,7 @@ export interface CodexAppServerPatchedProtocolOptions { readonly terminationError?: Effect.Effect; readonly logIncoming?: boolean; readonly logOutgoing?: boolean; + readonly maxIncomingMessageBytes?: number; readonly logger?: (event: CodexAppServerProtocolLogEvent) => Effect.Effect; readonly onNotification?: ( notification: CodexAppServerIncomingNotification, @@ -164,7 +183,9 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa yield* Queue.sliding(MAX_BUFFERED_RAW_MESSAGES); const pending = yield* Ref.make(new Map()); const nextRequestId = yield* Ref.make(1); + const maxIncomingMessageBytes = options.maxIncomingMessageBytes ?? MAX_INCOMING_MESSAGE_BYTES; const remainder: Array = []; + let remainderBytes = 0; const terminationHandled = yield* Ref.make(false); const terminationFailure = yield* Ref.make(Option.none()); const terminationSignal = yield* Deferred.make(); @@ -399,25 +420,49 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Stream.interruptWhen(Deferred.await(terminationSignal)), Stream.decodeText(), Stream.runForEach((chunk) => - Effect.sync(() => { + Effect.suspend(() => { const lines: Array = []; let start = 0; - for ( - let newline = chunk.indexOf("\n"); - newline !== -1; - newline = chunk.indexOf("\n", start) - ) { + while (start < chunk.length) { + const newline = chunk.indexOf("\n", start); + const end = newline === -1 ? chunk.length : newline; + const bytes = remainderBytes + utf8ByteLength(chunk, start, end); + // A \r ending the fragment may be half of a CRLF terminator, so it does + // not count yet. An empty fragment adds nothing to a remainder that fits. + const overLimit = + end > start && + bytes - (chunk.charCodeAt(end - 1) === 0x0d ? 1 : 0) > maxIncomingMessageBytes; + if (overLimit) { + remainder.length = 0; + remainderBytes = 0; + // Complete lines from earlier in this chunk are still handled, in order. + return Effect.forEach(lines, handleLine, { discard: true }).pipe( + Effect.andThen( + Effect.fail( + new CodexError.CodexAppServerTransportError({ + operation: "read-input-stream", + cause: new Error( + `Incoming message reached ${bytes} bytes, over the ${maxIncomingMessageBytes}-byte limit.`, + ), + }), + ), + ), + ); + } + if (newline === -1) { + // Keep unfinished lines in fragments so each chunk is scanned only once. + remainder.push(chunk.slice(start)); + remainderBytes = bytes; + break; + } remainder.push(chunk.slice(start, newline)); lines.push(remainder.join("").replace(/\r$/, "")); remainder.length = 0; + remainderBytes = 0; start = newline + 1; } - // Keep unfinished lines in fragments so each chunk is scanned only once. - if (start < chunk.length) { - remainder.push(chunk.slice(start)); - } - return lines; - }).pipe(Effect.flatMap((lines) => Effect.forEach(lines, handleLine, { discard: true }))), + return Effect.forEach(lines, handleLine, { discard: true }); + }), ), Effect.matchEffect({ onFailure: (error) => @@ -425,12 +470,24 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Effect.succeed(normalizeIncomingError(error, "read-input-stream")), ), onSuccess: () => - Effect.sync(() => { + Effect.suspend(() => { + // With no terminator left to strip, a final \r counts toward the limit. + if (remainderBytes > maxIncomingMessageBytes) { + const error = new CodexError.CodexAppServerTransportError({ + operation: "read-input-stream", + cause: new Error( + `Incoming message reached ${remainderBytes} bytes, over the ${maxIncomingMessageBytes}-byte limit.`, + ), + }); + remainder.length = 0; + remainderBytes = 0; + return Effect.fail(error); + } const line = remainder.join(""); remainder.length = 0; - return line; + remainderBytes = 0; + return handleLine(line); }).pipe( - Effect.flatMap(handleLine), Effect.matchEffect({ onFailure: (error) => handleTermination(() => Effect.succeed(error)), onSuccess: () => From cc044d72b2c5f06e24f3110ddbd7ba1c7434d4a6 Mon Sep 17 00:00:00 2001 From: Victor Vogel <263261067+only21mil@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:16:12 -0500 Subject: [PATCH 2/5] fix(codex): keep the unfinished-message buffer bounded when input arrives a byte at a time The reader kept one remainder entry per decoded chunk, so a peer that sends one byte per read could hold about 134 million entries, over 1 GiB of references, under the 128 MiB ceiling before the byte check tripped. Fragments shorter than 1024 UTF-16 code units now wait in a short tail that is joined into one entry once it reaches that length, and longer reads are kept as they are, so entries grow with bytes rather than reads. The byte check, CRLF handling and dispatch order are unchanged. Co-authored-by: Cestercian <183791452+cestercian@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- .../src/protocol.test.ts | 130 ++++++++++++++++++ .../effect-codex-app-server/src/protocol.ts | 53 +++++-- 2 files changed, 174 insertions(+), 9 deletions(-) diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 249924aa1f32c..17a1443b76d79 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -41,6 +41,16 @@ const lineFramings = [ { label: "CRLF split by an empty chunk", frame: (line: string) => [`${line}\r`, "", "\n"] }, ]; +// REMAINDER_BLOCK_LENGTH in protocol.ts: unfinished fragments shorter than this +// many UTF-16 code units are joined into one entry once they add up to it. +const remainderBlockLength = 1024; + +// Text that differs by position, so a fragment kept out of order changes it. +const positionalText = (length: number) => + Array.from({ length }, (_, index) => index.toString(36)) + .join("") + .slice(0, length); + const makeLimitedProtocol = (maxIncomingMessageBytes: number) => Effect.gen(function* () { const { stdio, input, output } = yield* makeInMemoryStdio(); @@ -446,6 +456,126 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); + it.effect.each([ + { label: "LF after several blocks", length: 5 * remainderBlockLength + 300, terminator: "\n" }, + { + label: "CRLF whose \\r completes a block", + length: 4 * remainderBlockLength - 1, + terminator: "\r\n", + }, + ])( + "routes a message sent one byte per chunk at the byte limit with $label", + ({ length, terminator }) => + Effect.gen(function* () { + const empty = encodeUnknownJsonString({ method: "x/bytes", params: { text: "" } }); + const message = { + method: "x/bytes", + params: { text: positionalText(length - empty.length) }, + }; + const line = encodeUnknownJsonString(message); + assert.lengthOf(line, length); + const { input, termination, notifications } = yield* makeLimitedProtocol(length); + + const bytes = encoder.encode(`${line}${terminator}`); + for (let offset = 0; offset < bytes.length; offset++) { + yield* Queue.offer(input, bytes.subarray(offset, offset + 1)); + } + yield* Queue.end(input); + + assert.instanceOf( + yield* Deferred.await(termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(notifications, [message]); + }), + ); + + it.effect("keeps one-byte chunks in order around a chunk longer than a block", () => + Effect.gen(function* () { + const message = { + method: "x/mixed", + params: { text: positionalText(4 * remainderBlockLength) }, + }; + const bytes = encodeJsonl(message); + const { input, termination, notifications } = yield* makeLimitedProtocol(bytes.length - 1); + + const longEnd = 100 + 2 * remainderBlockLength; + const chunks = [ + ...Array.from({ length: 100 }, (_, offset) => bytes.subarray(offset, offset + 1)), + bytes.subarray(100, longEnd), + ...Array.from({ length: 100 }, (_, index) => + bytes.subarray(longEnd + index, longEnd + index + 1), + ), + bytes.subarray(longEnd + 100), + ]; + for (const chunk of chunks) { + yield* Queue.offer(input, chunk); + } + yield* Queue.end(input); + + assert.instanceOf( + yield* Deferred.await(termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(notifications, [message]); + }), + ); + + it.effect( + "rejects an over-limit message sent one byte per chunk before its newline arrives", + () => + Effect.gen(function* () { + const limit = 3 * remainderBlockLength + 5; + const { input, output, transport, termination, notifications, events } = + yield* makeLimitedProtocol(limit); + const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped); + yield* Queue.take(output); + + const bytes = encoder.encode(`{"method":"x/huge","params":{"text":"${"x".repeat(limit)}`); + for (let offset = 0; offset < bytes.length; offset++) { + yield* Queue.offer(input, bytes.subarray(offset, offset + 1)); + } + + const error = yield* awaitMessageLimitFailure(termination); + const pendingError = yield* Fiber.join(pending).pipe( + Effect.match({ + onFailure: (failure) => failure, + onSuccess: () => assert.fail("Expected the oversized message to fail the request"), + }), + ); + assert.strictEqual(pendingError, error); + assert.deepEqual(notifications, []); + assert.deepEqual(events, []); + }), + ); + + it.effect.each([1, 7])( + "keeps multi-byte characters intact across remainder blocks with %i-byte input chunks", + (chunkSize) => + Effect.gen(function* () { + const head = '{"method":"x/unicode","params":{"text":"'; + // The emoji's surrogate pair straddles the first block edge; 3-byte + // characters fill the blocks after it. + const text = `${"Γ©".repeat(remainderBlockLength - 1 - head.length)}πŸ™‚${"ζœ€".repeat(3 * remainderBlockLength)}`; + const message = { method: "x/unicode", params: { text } }; + const line = encodeUnknownJsonString(message); + assert.strictEqual(line.indexOf("πŸ™‚"), remainderBlockLength - 1); + const bytes = encoder.encode(`${line}\n`); + const { input, termination, notifications } = yield* makeLimitedProtocol(bytes.length - 1); + + for (let offset = 0; offset < bytes.length; offset += chunkSize) { + yield* Queue.offer(input, bytes.subarray(offset, offset + chunkSize)); + } + yield* Queue.end(input); + + assert.instanceOf( + yield* Deferred.await(termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(notifications, [message]); + }), + ); + it.effect("routes an unterminated final message at the byte limit when input ends", () => Effect.gen(function* () { const message = { method: "x/final", params: { text: "no newline" } }; diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 97adb1511aee7..56475e7a24a5c 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -20,6 +20,11 @@ const MAX_BUFFERED_RAW_MESSAGES = 32; // Codex diffs reach ~49M characters and grow when JSON-escaped; a line far // past this ceiling would exhaust the heap during join and parse. const MAX_INCOMING_MESSAGE_BYTES = 128 * 1024 * 1024; +// Unfinished-message fragments shorter than this many UTF-16 code units are +// joined into one entry once they add up to it, so a peer that sends one byte +// per read holds one entry per block rather than one per byte. Longer reads, +// such as 64 KiB pipe reads, are kept as they are without a copy. +const REMAINDER_BLOCK_LENGTH = 1024; // UTF-8 size of `text.slice(start, end)` without encoding a copy. Text from a // streaming TextDecoder holds no lone surrogates, so each surrogate is one half @@ -185,6 +190,8 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa const nextRequestId = yield* Ref.make(1); const maxIncomingMessageBytes = options.maxIncomingMessageBytes ?? MAX_INCOMING_MESSAGE_BYTES; const remainder: Array = []; + const remainderTail: Array = []; + let remainderTailLength = 0; let remainderBytes = 0; const terminationHandled = yield* Ref.make(false); const terminationFailure = yield* Ref.make(Option.none()); @@ -416,6 +423,36 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa ); }; + // Fragments shorter than a block wait in the tail, which stays under a block + // of code units and so under that many entries. Packing joins the tail into + // one remainder entry, so a short fragment is copied here and once more when + // its line is joined, never again. + const packRemainderTail = () => { + if (remainderTail.length > 0) { + remainder.push(remainderTail.join("")); + remainderTail.length = 0; + remainderTailLength = 0; + } + }; + const retainFragment = (fragment: string) => { + if (fragment.length >= REMAINDER_BLOCK_LENGTH) { + packRemainderTail(); + remainder.push(fragment); + return; + } + remainderTail.push(fragment); + remainderTailLength += fragment.length; + if (remainderTailLength >= REMAINDER_BLOCK_LENGTH) { + packRemainderTail(); + } + }; + const clearRemainder = () => { + remainder.length = 0; + remainderTail.length = 0; + remainderTailLength = 0; + remainderBytes = 0; + }; + yield* options.stdio.stdin.pipe( Stream.interruptWhen(Deferred.await(terminationSignal)), Stream.decodeText(), @@ -433,8 +470,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa end > start && bytes - (chunk.charCodeAt(end - 1) === 0x0d ? 1 : 0) > maxIncomingMessageBytes; if (overLimit) { - remainder.length = 0; - remainderBytes = 0; + clearRemainder(); // Complete lines from earlier in this chunk are still handled, in order. return Effect.forEach(lines, handleLine, { discard: true }).pipe( Effect.andThen( @@ -451,14 +487,14 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa } if (newline === -1) { // Keep unfinished lines in fragments so each chunk is scanned only once. - remainder.push(chunk.slice(start)); + retainFragment(chunk.slice(start)); remainderBytes = bytes; break; } + packRemainderTail(); remainder.push(chunk.slice(start, newline)); lines.push(remainder.join("").replace(/\r$/, "")); - remainder.length = 0; - remainderBytes = 0; + clearRemainder(); start = newline + 1; } return Effect.forEach(lines, handleLine, { discard: true }); @@ -479,13 +515,12 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa `Incoming message reached ${remainderBytes} bytes, over the ${maxIncomingMessageBytes}-byte limit.`, ), }); - remainder.length = 0; - remainderBytes = 0; + clearRemainder(); return Effect.fail(error); } + packRemainderTail(); const line = remainder.join(""); - remainder.length = 0; - remainderBytes = 0; + clearRemainder(); return handleLine(line); }).pipe( Effect.matchEffect({ From 0c53c13dd081f51c35ab84bfee3cff162c60d167 Mon Sep 17 00:00:00 2001 From: Victor Vogel <263261067+only21mil@users.noreply.github.com> Date: Sun, 11 Oct 2026 11:16:13 -0500 Subject: [PATCH 3/5] fix(codex): copy a fragment that follows a newline so it does not keep the whole read alive V8 keeps a slice of 13 or more characters as a view into its parent string, and a one-item join returns that view. So a short unfinished fragment at the end of a large read kept the whole decoded read alive: a 64-byte prefix after 64 MiB of completed lines held about 64 MiB while the byte count said 2,112. A fragment that follows a newline is now copied with a two-item join, which builds a new flat string. Only a message's first fragment can follow a newline, so each message pays for at most one copy of at most one read, and whole reads are still kept without a copy. Co-authored-by: Cestercian <183791452+cestercian@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- .../src/protocol.test.ts | 35 +++++++++++++++++++ .../effect-codex-app-server/src/protocol.ts | 18 +++++++--- 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 17a1443b76d79..5ae5c46dba30b 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -576,6 +576,41 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); + it.effect.each([ + { label: "short", prefixLength: 64 }, + { label: "long", prefixLength: remainderBlockLength + 76 }, + ])( + "routes a message with a $label first fragment after complete lines in a large chunk", + ({ prefixLength }) => + Effect.gen(function* () { + const before = { method: "x/before", params: { index: 1 } }; + const message = { + method: "x/after", + params: { text: `ζœ€πŸ™‚${positionalText(4 * remainderBlockLength)}` }, + }; + const line = encodeUnknownJsonString(message); + const lineLimit = 64 * 1024; + const { input, termination, notifications } = yield* makeLimitedProtocol(lineLimit); + + // Whitespace lines make the chunk far longer than the fragment kept from its end. + const blankLines = `${" ".repeat(lineLimit)}\n`.repeat(16); + yield* Queue.offer( + input, + encoder.encode( + `${blankLines}${encodeUnknownJsonString(before)}\n${line.slice(0, prefixLength)}`, + ), + ); + yield* Queue.offer(input, encoder.encode(`${line.slice(prefixLength)}\n`)); + yield* Queue.end(input); + + assert.instanceOf( + yield* Deferred.await(termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(notifications, [before, message]); + }), + ); + it.effect("routes an unterminated final message at the byte limit when input ends", () => Effect.gen(function* () { const message = { method: "x/final", params: { text: "no newline" } }; diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 56475e7a24a5c..3cc69dcb1a12b 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -22,10 +22,17 @@ const MAX_BUFFERED_RAW_MESSAGES = 32; const MAX_INCOMING_MESSAGE_BYTES = 128 * 1024 * 1024; // Unfinished-message fragments shorter than this many UTF-16 code units are // joined into one entry once they add up to it, so a peer that sends one byte -// per read holds one entry per block rather than one per byte. Longer reads, -// such as 64 KiB pipe reads, are kept as they are without a copy. +// per read holds one entry per block rather than one per byte. A whole longer +// read, such as a 64 KiB pipe read, is kept as it is without a copy. const REMAINDER_BLOCK_LENGTH = 1024; +// `text.slice(start)` as a new string that does not keep `text` alive. V8 +// returns a slice of 13 or more code units, and a one-item join of that slice, +// as a view that holds the whole of `text`; a join of two items builds a new +// flat string. +const sliceDetached = (text: string, start: number) => + [text.slice(start, start + 1), text.slice(start + 1)].join(""); + // UTF-8 size of `text.slice(start, end)` without encoding a copy. Text from a // streaming TextDecoder holds no lone surrogates, so each surrogate is one half // of a four-byte sequence. @@ -426,7 +433,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa // Fragments shorter than a block wait in the tail, which stays under a block // of code units and so under that many entries. Packing joins the tail into // one remainder entry, so a short fragment is copied here and once more when - // its line is joined, never again. + // its line is joined, plus once on arrival if it follows a newline. const packRemainderTail = () => { if (remainderTail.length > 0) { remainder.push(remainderTail.join("")); @@ -487,7 +494,10 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa } if (newline === -1) { // Keep unfinished lines in fragments so each chunk is scanned only once. - retainFragment(chunk.slice(start)); + // A fragment after a newline is copied so it does not keep the lines + // before it alive. Only a message's first fragment can follow a + // newline, so a message pays for at most one such copy of one read. + retainFragment(start === 0 ? chunk : sliceDetached(chunk, start)); remainderBytes = bytes; break; } From a3c988e183eb0af5acce19213e1f1a80b80d9484 Mon Sep 17 00:00:00 2001 From: Victor Vogel <263261067+only21mil@users.noreply.github.com> Date: Sun, 11 Oct 2026 20:14:34 -0500 Subject: [PATCH 4/5] test(codex): bound the open-input limit tests and cover U+FFFD counting The two tests that keep input open now fail in 5 s instead of the 60 s runner default. A new test sends an invalid UTF-8 byte and checks that its U+FFFD replacement counts three bytes. The log test also checks that the error cause names the limit and the bytes seen. Co-Authored-By: Claude Opus 5.5 --- .../src/protocol.test.ts | 84 ++++++++++++++----- 1 file changed, 62 insertions(+), 22 deletions(-) diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 5ae5c46dba30b..be538199457ad 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -431,29 +431,32 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); - it.effect("rejects an over-limit message from small fragments before its newline arrives", () => - Effect.gen(function* () { - const { input, output, transport, termination, notifications, events } = - yield* makeLimitedProtocol(1024); - const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped); - yield* Queue.take(output); + it.effect( + "rejects an over-limit message from small fragments before its newline arrives", + () => + Effect.gen(function* () { + const { input, output, transport, termination, notifications, events } = + yield* makeLimitedProtocol(1024); + const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped); + yield* Queue.take(output); - yield* Queue.offer(input, encoder.encode('{"method":"x/huge","params":{"text":"')); - for (let index = 0; index < 512; index++) { - yield* Queue.offer(input, encoder.encode("xxxxxxxx")); - } + yield* Queue.offer(input, encoder.encode('{"method":"x/huge","params":{"text":"')); + for (let index = 0; index < 512; index++) { + yield* Queue.offer(input, encoder.encode("xxxxxxxx")); + } - const error = yield* awaitMessageLimitFailure(termination); - const pendingError = yield* Fiber.join(pending).pipe( - Effect.match({ - onFailure: (failure) => failure, - onSuccess: () => assert.fail("Expected the oversized message to fail the request"), - }), - ); - assert.strictEqual(pendingError, error); - assert.deepEqual(notifications, []); - assert.deepEqual(events, []); - }), + const error = yield* awaitMessageLimitFailure(termination); + const pendingError = yield* Fiber.join(pending).pipe( + Effect.match({ + onFailure: (failure) => failure, + onSuccess: () => assert.fail("Expected the oversized message to fail the request"), + }), + ); + assert.strictEqual(pendingError, error); + assert.deepEqual(notifications, []); + assert.deepEqual(events, []); + }), + 5_000, ); it.effect.each([ @@ -547,6 +550,7 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { assert.deepEqual(notifications, []); assert.deepEqual(events, []); }), + 5_000, ); it.effect.each([1, 7])( @@ -680,14 +684,48 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); + it.effect("counts an invalid UTF-8 byte as the three bytes of its U+FFFD replacement", () => + Effect.gen(function* () { + const message = { method: "x/invalid", params: { text: "a\uFFFDb" } }; + const line = encodeUnknownJsonString(message); + const lineBytes = encoder.encode(line).length; + // 0xff is never valid UTF-8, so the decoder replaces it with U+FFFD. + const [head, tail] = line.split("\uFFFD"); + const bytes = Uint8Array.from([ + ...encoder.encode(head), + 0xff, + ...encoder.encode(`${tail}\n`), + ]); + // Counting raw input bytes instead would accept the over-limit message. + assert.strictEqual(bytes.length - 1, lineBytes - 2); + + const atLimit = yield* makeLimitedProtocol(lineBytes); + const overLimit = yield* makeLimitedProtocol(lineBytes - 1); + yield* Queue.offer(atLimit.input, bytes); + yield* Queue.offer(overLimit.input, bytes); + yield* Queue.end(atLimit.input); + yield* Queue.end(overLimit.input); + + assert.instanceOf( + yield* Deferred.await(atLimit.termination), + CodexError.CodexAppServerInputStreamEndedError, + ); + assert.deepEqual(atLimit.notifications, [message]); + yield* awaitMessageLimitFailure(overLimit.termination); + assert.deepEqual(overLimit.notifications, []); + assert.deepEqual(overLimit.events, []); + }), + ); + it.effect("keeps an over-limit message out of protocol logs and the termination error", () => Effect.gen(function* () { const secret = "codex-oversized-secret-sentinel"; const small = { method: "x/small", params: {} }; + const huge = encodeJsonl({ method: "x/huge", params: { text: secret } }); const { input, termination, notifications, events } = yield* makeLimitedProtocol(64); yield* Queue.offer(input, encodeJsonl(small)); - yield* Queue.offer(input, encodeJsonl({ method: "x/huge", params: { text: secret } })); + yield* Queue.offer(input, huge); yield* Queue.end(input); const error = yield* awaitMessageLimitFailure(termination); @@ -699,6 +737,8 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { assert.notInclude(encodeUnknownJsonString(events), secret); assert.notInclude(error.message, secret); assert.notInclude(String(error.cause), secret); + assert.include(String(error.cause), `reached ${huge.length - 1} bytes`); + assert.include(String(error.cause), "over the 64-byte limit"); }), ); From b35ffeff8eff88bc9a541e17c90fa89b59b3c683 Mon Sep 17 00:00:00 2001 From: Victor Vogel <263261067+only21mil@users.noreply.github.com> Date: Sun, 11 Oct 2026 20:15:30 -0500 Subject: [PATCH 5/5] docs(codex): mark the incoming message byte limit option as a test override Co-Authored-By: Claude Opus 5.5 --- packages/effect-codex-app-server/src/protocol.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 3cc69dcb1a12b..7d3b05e0f5dcd 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -69,6 +69,7 @@ export interface CodexAppServerPatchedProtocolOptions { readonly terminationError?: Effect.Effect; readonly logIncoming?: boolean; readonly logOutgoing?: boolean; + // Test override; production uses the 128 MiB default. readonly maxIncomingMessageBytes?: number; readonly logger?: (event: CodexAppServerProtocolLogEvent) => Effect.Effect; readonly onNotification?: (