fix(codex): parse jsonl frames linearly

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

packages/effect-codex-app-server/src/protocol.ts

@@ -157,7 +157,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
     const incomingRequests = yield* Queue.unbounded<CodexAppServerIncomingRequest>();
     const pending = yield* Ref.make(new Map<string, CodexAppServerPendingRequest>());
     const nextRequestId = yield* Ref.make(1);
-    const remainder = yield* Ref.make("");
+    const pendingLineParts: Array<string> = [];
     const terminationHandled = yield* Ref.make(false);
     const scope = yield* Effect.scope;
 
@@ -358,20 +358,37 @@ 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) =>
           handleTermination(() =>
             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)),