From 6895996ded1653484e0fab46f8dc4d8c110f281c Mon Sep 17 00:00:00 2001 From: Taras Date: Thu, 27 Aug 2026 20:45:24 +0300 Subject: [PATCH 1/4] fix(codex): parse jsonl frames linearly --- .../effect-codex-app-server/src/protocol.ts | 33 ++++++++++++++----- 1 file changed, 25 insertions(+), 8 deletions(-) diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 17bfaed2b64c..32cf70405c41 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -157,7 +157,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa const incomingRequests = yield* Queue.unbounded(); const pending = yield* Ref.make(new Map()); const nextRequestId = yield* Ref.make(1); - const remainder = yield* Ref.make(""); + const pendingLineParts: Array = []; const terminationHandled = yield* Ref.make(false); const logProtocol = (event: CodexAppServerProtocolLogEvent) => { @@ -354,12 +354,29 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa yield* options.stdio.stdin.pipe( Stream.decodeText(), Stream.runForEach((chunk) => - Ref.modify(remainder, (current) => { - const combined = current + chunk; - const lines = combined.split("\n"); - const nextRemainder = lines.pop() ?? ""; - return [lines.map((line) => line.replace(/\r$/, "")), nextRemainder] as const; - }).pipe(Effect.flatMap((lines) => Effect.forEach(lines, handleLine, { discard: true }))), + Effect.gen(function* () { + let start = 0; + let newline = chunk.indexOf("\n"); + + while (newline !== -1) { + let line = chunk.slice(start, newline); + if (pendingLineParts.length > 0) { + pendingLineParts.push(line); + line = pendingLineParts.join(""); + pendingLineParts.length = 0; + } + if (line.endsWith("\r")) { + line = line.slice(0, -1); + } + yield* handleLine(line); + start = newline + 1; + newline = chunk.indexOf("\n", start); + } + + if (start < chunk.length) { + pendingLineParts.push(chunk.slice(start)); + } + }), ), Effect.matchEffect({ onFailure: (error) => @@ -367,7 +384,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Effect.succeed(normalizeIncomingError(error, "read-input-stream")), ), onSuccess: () => - Ref.get(remainder).pipe( + Effect.sync(() => pendingLineParts.join("")).pipe( Effect.flatMap((line) => (line.trim().length === 0 ? Effect.void : handleLine(line))), Effect.matchEffect({ onFailure: (error) => handleTermination(() => Effect.succeed(error)), From 2ad52e272d73903907d87b7a78ee6e78b041e695 Mon Sep 17 00:00:00 2001 From: Taras Date: Thu, 27 Aug 2026 21:24:46 +0300 Subject: [PATCH 2/4] test(codex): cover fragmented jsonl frames --- .../src/protocol.test.ts | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index a7e0397b4adb..802bb8360287 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -290,6 +290,64 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); + it.effect("routes fragmented and coalesced JSONL frames", () => + Effect.gen(function* () { + const { stdio, input } = yield* makeInMemoryStdio(); + const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio }); + const notifications = yield* transport.incomingNotifications.pipe( + Stream.take(3), + Stream.runCollect, + Effect.forkScoped, + ); + + const fragmentedNotification = { + method: "item/agentMessage/delta", + params: { + delta: "fragmented", + itemId: "item-1", + threadId: "thread-1", + turnId: "turn-1", + }, + }; + const fragmentedFrame = encodeJsonl(fragmentedNotification); + yield* Queue.offer(input, fragmentedFrame.slice(0, 13)); + yield* Queue.offer(input, fragmentedFrame.slice(13, fragmentedFrame.length - 1)); + yield* Queue.offer(input, fragmentedFrame.slice(fragmentedFrame.length - 1)); + + const coalescedNotifications = [ + { + method: "item/agentMessage/delta", + params: { + delta: "coalesced-1", + itemId: "item-2", + threadId: "thread-1", + turnId: "turn-1", + }, + }, + { + method: "item/agentMessage/delta", + params: { + delta: "coalesced-2", + itemId: "item-3", + threadId: "thread-1", + turnId: "turn-1", + }, + }, + ]; + yield* Queue.offer( + input, + encoder.encode( + `${encodeUnknownJsonString(coalescedNotifications[0])}\n${encodeUnknownJsonString(coalescedNotifications[1])}\n`, + ), + ); + + assert.deepEqual(yield* Fiber.join(notifications), [ + fragmentedNotification, + ...coalescedNotifications, + ]); + }), + ); + it.effect("surfaces JSON encoding failures as protocol parse errors", () => Effect.gen(function* () { const { stdio } = yield* makeInMemoryStdio(); From 02f6a3e89c9fe9dc7d4e82c22925deb68a80a7f8 Mon Sep 17 00:00:00 2001 From: Taras Date: Thu, 27 Aug 2026 21:51:08 +0300 Subject: [PATCH 3/4] refactor(codex): use effect stream line splitting --- .../effect-codex-app-server/src/protocol.ts | 43 +++---------------- 1 file changed, 6 insertions(+), 37 deletions(-) diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 32cf70405c41..e87f5e9d8ae4 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -157,7 +157,6 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa const incomingRequests = yield* Queue.unbounded(); const pending = yield* Ref.make(new Map()); const nextRequestId = yield* Ref.make(1); - const pendingLineParts: Array = []; const terminationHandled = yield* Ref.make(false); const logProtocol = (event: CodexAppServerProtocolLogEvent) => { @@ -353,48 +352,18 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa yield* options.stdio.stdin.pipe( Stream.decodeText(), - Stream.runForEach((chunk) => - Effect.gen(function* () { - let start = 0; - let newline = chunk.indexOf("\n"); - - while (newline !== -1) { - let line = chunk.slice(start, newline); - if (pendingLineParts.length > 0) { - pendingLineParts.push(line); - line = pendingLineParts.join(""); - pendingLineParts.length = 0; - } - if (line.endsWith("\r")) { - line = line.slice(0, -1); - } - yield* handleLine(line); - start = newline + 1; - newline = chunk.indexOf("\n", start); - } - - if (start < chunk.length) { - pendingLineParts.push(chunk.slice(start)); - } - }), - ), + Stream.splitLines, + Stream.runForEach(handleLine), Effect.matchEffect({ onFailure: (error) => handleTermination(() => Effect.succeed(normalizeIncomingError(error, "read-input-stream")), ), onSuccess: () => - Effect.sync(() => pendingLineParts.join("")).pipe( - Effect.flatMap((line) => (line.trim().length === 0 ? Effect.void : handleLine(line))), - Effect.matchEffect({ - onFailure: (error) => handleTermination(() => Effect.succeed(error)), - onSuccess: () => - handleTermination( - () => - options.terminationError ?? - Effect.succeed(new CodexError.CodexAppServerInputStreamEndedError({})), - ), - }), + handleTermination( + () => + options.terminationError ?? + Effect.succeed(new CodexError.CodexAppServerInputStreamEndedError({})), ), }), Effect.forkScoped, From 8c1db600ee77a767f4131c129e302a1556ca4867 Mon Sep 17 00:00:00 2001 From: Taras Date: Thu, 27 Aug 2026 21:55:44 +0300 Subject: [PATCH 4/4] test(codex): remove jsonl framing regression --- .../src/protocol.test.ts | 58 ------------------- 1 file changed, 58 deletions(-) diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 802bb8360287..a7e0397b4adb 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -290,64 +290,6 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); - it.effect("routes fragmented and coalesced JSONL frames", () => - Effect.gen(function* () { - const { stdio, input } = yield* makeInMemoryStdio(); - const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ stdio }); - const notifications = yield* transport.incomingNotifications.pipe( - Stream.take(3), - Stream.runCollect, - Effect.forkScoped, - ); - - const fragmentedNotification = { - method: "item/agentMessage/delta", - params: { - delta: "fragmented", - itemId: "item-1", - threadId: "thread-1", - turnId: "turn-1", - }, - }; - const fragmentedFrame = encodeJsonl(fragmentedNotification); - yield* Queue.offer(input, fragmentedFrame.slice(0, 13)); - yield* Queue.offer(input, fragmentedFrame.slice(13, fragmentedFrame.length - 1)); - yield* Queue.offer(input, fragmentedFrame.slice(fragmentedFrame.length - 1)); - - const coalescedNotifications = [ - { - method: "item/agentMessage/delta", - params: { - delta: "coalesced-1", - itemId: "item-2", - threadId: "thread-1", - turnId: "turn-1", - }, - }, - { - method: "item/agentMessage/delta", - params: { - delta: "coalesced-2", - itemId: "item-3", - threadId: "thread-1", - turnId: "turn-1", - }, - }, - ]; - yield* Queue.offer( - input, - encoder.encode( - `${encodeUnknownJsonString(coalescedNotifications[0])}\n${encodeUnknownJsonString(coalescedNotifications[1])}\n`, - ), - ); - - assert.deepEqual(yield* Fiber.join(notifications), [ - fragmentedNotification, - ...coalescedNotifications, - ]); - }), - ); - it.effect("surfaces JSON encoding failures as protocol parse errors", () => Effect.gen(function* () { const { stdio } = yield* makeInMemoryStdio();