Commit 702a6ade3c69bf1c9ce06ad0451dbd8e0ecc7be9 | the-knife

author	Julius Marminge <julius0216@outlook.com> 2026-08-28 18:17:06-07:00
committer	GitHub <noreply@github.com> 2026-08-28 18:17:06-07:00
log	702a6ade3c69bf1c9ce06ad0451dbd8e0ecc7be9
tree	69082af2bb5c02346dd3bc82e1e818009b8113de
parent	2bc9e8ef6ad57a2513e4552e8a0b21d53db859e6
signature	Signed by PGP key B5690EEEBB952194

fix(codex): avoid quadratic app-server input buffering (#8605)

2 files changed, 133 insertions(+), 8 deletions(-)

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

packages/effect-codex-app-server/src/protocol.test.ts+108

...	@@ -290,6 +290,114 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {	...	@@ -290,6 +290,114 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
290	}),	290	}),
291	);	291	);
292		292
		293	it.effect("routes a large notification fragmented across thousands of input chunks", () =>
		294	Effect.gen(function* () {
		295	const { stdio, input, output } = yield* makeInMemoryStdio();
		296	const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];
		297	const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
		298	stdio,
		299	onNotification: (notification) =>
		300	Effect.sync(() => {
		301	notifications.push(notification);
		302	}),
		303	});
		304	const response = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
		305	yield* Queue.take(output);
		306
		307	const notification = {
		308	method: "turn/diff/updated",
		309	params: {
		310	threadId: "thread-1",
		311	turnId: "turn-1",
		312	diff: "x".repeat(4 * 1024 * 1024),
		313	},
		314	};
		315	const bytes = encoder.encode(
		316	`${encodeUnknownJsonString(notification)}\n${encodeUnknownJsonString({ id: 1, result: { ok: true } })}\n`,
		317	);
		318	for (let offset = 0; offset < bytes.length; offset += 1024) {
		319	yield* Queue.offer(input, bytes.subarray(offset, offset + 1024));
		320	}
		321
		322	assert.deepEqual(yield* Fiber.join(response), { ok: true });
		323	assert.deepEqual(notifications, [notification]);
		324	}),
		325	);
		326
		327	it.effect.each([1, 7, 1024])(
		328	"preserves JSONL framing and UTF-8 across %i-byte input chunks",
		329	(chunkSize) =>
		330	Effect.gen(function* () {
		331	const { stdio, input } = yield* makeInMemoryStdio();
		332	const notifications: Array<CodexProtocol.CodexAppServerIncomingNotification> = [];
		333	const rawLines: Array<unknown> = [];
		334	const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
		335	yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
		336	stdio,
		337	logIncoming: true,
		338	logger: (event) =>
		339	Effect.sync(() => {
		340	if (event.stage === "raw") {
		341	rawLines.push(event.payload);
		342	}
		343	}),
		344	onNotification: (notification) =>
		345	Effect.sync(() => {
		346	notifications.push(notification);
		347	}),
		348	onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
		349	});
		350
		351	const firstLine = '{"method":"x/first",\r"params":{"text":"hé🙂"}}';
		352	const secondLine = '{"method":"x/second","params":{"value":2}}';
		353	const finalLine = '{"method":"x/final","params":{"text":"最後"}}\r';
		354	const bytes = encoder.encode(`\n \t\r\n${firstLine}\r\n\n${secondLine}\n${finalLine}`);
		355	for (let offset = 0; offset < bytes.length; offset += chunkSize) {
		356	yield* Queue.offer(input, bytes.subarray(offset, offset + chunkSize));
		357	}
		358	yield* Queue.end(input);
		359
		360	assert.instanceOf(
		361	yield* Deferred.await(termination),
		362	CodexError.CodexAppServerInputStreamEndedError,
		363	);
		364	assert.deepEqual(notifications, [
		365	{ method: "x/first", params: { text: "hé🙂" } },
		366	{ method: "x/second", params: { value: 2 } },
		367	{ method: "x/final", params: { text: "最後" } },
		368	]);
		369	assert.deepEqual(rawLines, [firstLine, secondLine, finalLine]);
		370	}),
		371	);
		372
		373	it.effect("reports a malformed fragmented final line before input stream termination", () =>
		374	Effect.gen(function* () {
		375	const { stdio, input, output } = yield* makeInMemoryStdio();
		376	const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
		377	const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
		378	stdio,
		379	onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
		380	});
		381	const response = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped);
		382	yield* Queue.take(output);
		383
		384	yield* Queue.offer(input, encoder.encode('{"id":1,'));
		385	yield* Queue.offer(input, encoder.encode('"result":'));
		386	yield* Queue.end(input);
		387
		388	const error = yield* Deferred.await(termination);
		389	assert.instanceOf(error, CodexError.CodexAppServerProtocolParseError);
		390	assert.equal(error.operation, "decode-wire-message");
		391	const responseError = yield* Fiber.join(response).pipe(
		392	Effect.match({
		393	onFailure: (failure) => failure,
		394	onSuccess: () => assert.fail("Expected the malformed response to fail the request"),
		395	}),
		396	);
		397	assert.strictEqual(responseError, error);
		398	}),
		399	);
		400
293	it.effect("surfaces JSON encoding failures as protocol parse errors", () =>	401	it.effect("surfaces JSON encoding failures as protocol parse errors", () =>
294	Effect.gen(function* () {	402	Effect.gen(function* () {
295	const { stdio } = yield* makeInMemoryStdio();	403	const { stdio } = yield* makeInMemoryStdio();

packages/effect-codex-app-server/src/protocol.ts+25-8

 @@ -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 remainder: Array<string> = [];
     const terminationHandled = yield* Ref.make(false);
     const logProtocol = (event: CodexAppServerProtocolLogEvent) => {
 @@ -354,11 +354,24 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
     yield* options.stdio.stdin.pipe(
       Stream.decodeText(),
       Stream.runForEach((chunk) =>
-        Ref.modify(remainder, (current) => {
+        Effect.sync(() => {
-          const combined = current + chunk;
+          const lines: Array<string> = [];
-          const lines = combined.split("\n");
+          let start = 0;
-          const nextRemainder = lines.pop() ?? "";
+          for (
-          return [lines.map((line) => line.replace(/\r$/, "")), nextRemainder] as const;
+            let newline = chunk.indexOf("\n");
+            newline !== -1;
+            newline = chunk.indexOf("\n", start)
+          ) {
+            remainder.push(chunk.slice(start, newline));
+            lines.push(remainder.join("").replace(/\r$/, ""));
+            remainder.length = 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 }))),
       ),
       Effect.matchEffect({
 @@ -367,8 +380,12 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
             Effect.succeed(normalizeIncomingError(error, "read-input-stream")),
           ),
         onSuccess: () =>
-          Ref.get(remainder).pipe(
+          Effect.sync(() => {
-            Effect.flatMap((line) => (line.trim().length === 0 ? Effect.void : handleLine(line))),
+            const line = remainder.join("");
+            remainder.length = 0;
+            return line;
+          }).pipe(
+            Effect.flatMap(handleLine),
             Effect.matchEffect({
               onFailure: (error) => handleTermination(() => Effect.succeed(error)),
               onSuccess: () =>