perf(command-code): measure fragmented NDJSON incrementally

src/adapters/command-code.ts

@@ -437,17 +437,20 @@ async function*ndjson(response: Response, budget: TranslatorBudget): AsyncGenera
   try {
     for (;;) {
       const { value, done } = await reader.read();
-      const next = buffer + decoder.decode(value, { stream: !done });
-      const nextBytes = encoder.encode(next).byteLength;
+      const decoded = decoder.decode(value, { stream: !done });
+      // TextDecoder emits complete code points, so fragment UTF-8 sizes are additive.
+      const nextBytes = bufferBytes + encoder.encode(decoded).byteLength;
       resizeBuffer(nextBytes);
-      buffer = next;
-      let newline = buffer.indexOf("\n");
+      buffer += decoded;
+      // The retained suffix has no newline; only new text can complete a record.
+      const hasNewline = decoded.includes("\n");
+      let newline = hasNewline ? buffer.indexOf("\n") : -1;
       while (newline >= 0) {
         const line = buffer.slice(0, newline).trim(); buffer = buffer.slice(newline + 1);
         if (line) yield* decodeEventLine(line);
         newline = buffer.indexOf("\n");
       }
-      resizeBuffer(encoder.encode(buffer).byteLength);
+      if (hasNewline) resizeBuffer(encoder.encode(buffer).byteLength);
       if (done) break;
     }
     const final = buffer.trim();

structure/transports/byte-accounting.md

@@ -155,7 +155,10 @@ removed bytes. Recounting an unchanged residual does not reserve it twice. Faile
 the previous charge owned until `finally` releases it, without releasing other owners' bytes.
 The turn limit and transport-read accounting remain in force. `tests/providers/command-code-buffer-budget.test.ts`
 covers exact-limit Unicode records across whole/one-byte reads and newline/EOF termination,
-one-byte overflow, another retained owner, consumer return and upstream cancellation.
+one-byte overflow, another retained owner, consumer return and upstream cancellation. Byte sizing
+measures only new decoded text, including EOF replacement bytes; newline searches run only when
+new text contains a newline, and only a drained residual is remeasured. The same tests bound total
+measurement and search input by wire size for both newline and EOF termination.
 
 Devin Connect-RPC in `src/adapters/devin/cloud-direct/chat.ts` compacts surviving chunks in place
 and truncates the queue, or clears it when no bytes remain. No deleted-chunk

tests/providers/command-code-buffer-budget.test.ts

@@ -1,4 +1,4 @@
-import { expect, test } from "bun:test";
+import { expect, spyOn, test } from "bun:test";
 import { createCommandCodeAdapter } from "../../src/adapters/command-code";
 import { createTestTranslatorBudget } from "../helpers/translator-budget";
 import type { AdapterEvent } from "../../src/types";
@@ -8,6 +8,41 @@ const content = "é한😀".repeat(32);
 const line = JSON.stringify({ type: "text-delta", text: content });
 const encoder = new TextEncoder();
 
+for (const ending of ["\n", ""] as const) {
+  test(`Command Code bounds fragmented NDJSON measurement and scan work with ${JSON.stringify(ending)} termination`, async () => {
+    const text = "a".repeat(256 * 256);
+    const bytes = encoder.encode(JSON.stringify({ type: "text-delta", text }) + ending);
+    const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length });
+    const source = input(bytes, 256);
+    const encode = TextEncoder.prototype.encode;
+    const indexOf = String.prototype.indexOf;
+    let measuredChars = 0;
+    let scannedChars = 0;
+    const measuring = spyOn(TextEncoder.prototype, "encode").mockImplementation(function (value = "") {
+      measuredChars += value.length;
+      return Reflect.apply(encode, this, [value]);
+    });
+    const scanning = spyOn(String.prototype, "indexOf").mockImplementation(function (search, position = 0) {
+      if (search === "\n") scannedChars += Math.max(0, this.length - position);
+      return Reflect.apply(indexOf, this, [search, position]);
+    });
+    const events: AdapterEvent[] = [];
+    try {
+      for await (const event of adapter().parseStream(source.response, budget)) events.push(event);
+    } finally {
+      measuring.mockRestore();
+      scanning.mockRestore();
+    }
+    expect(events).toEqual([
+      { type: "text_delta", text },
+      { type: "done", usage: undefined, stopReason: undefined },
+    ]);
+    expect(budget.snapshot()).toMatchObject({ currentBytes: 0, highWaterBytes: bytes.length, overflows: 0 });
+    expect(measuredChars).toBeLessThanOrEqual(4 * bytes.length);
+    expect(scannedChars).toBeLessThanOrEqual(2 * bytes.length);
+  });
+}
+
 function input(bytes: Uint8Array, chunkSize: number, beforeRead?: (offset: number) => void) {
   let offset = 0;
   let cancelled = false;
@@ -24,6 +59,26 @@ function input(bytes: Uint8Array, chunkSize: number, beforeRead?: (offset: numbe
   return { response: new Response(body), cancelled: () => cancelled };
 }
 
+for (const maxTurnBytes of [2, 3]) {
+  test(`Command Code accounts for EOF decoder replacement bytes with a ${maxTurnBytes}-byte budget`, async () => {
+    const budget = createTestTranslatorBudget({ maxTurnBytes });
+    const source = input(Uint8Array.of(0xe2), 1); // Incomplete UTF-8 becomes a 3-byte U+FFFD at EOF.
+    const events: AdapterEvent[] = [];
+    const consume = async () => {
+      for await (const event of adapter().parseStream(source.response, budget)) events.push(event);
+    };
+    if (maxTurnBytes === 2) {
+      await expect(consume()).rejects.toMatchObject({ code: "translation_buffer_limit", kind: "live_transient" });
+      expect(events).toEqual([]);
+    } else {
+      await consume();
+      // Malformed NDJSON retains the existing skip-and-complete behavior.
+      expect(events).toEqual([{ type: "done", usage: undefined, stopReason: undefined }]);
+    }
+    expect(budget.snapshot()).toMatchObject({ currentBytes: 0, overflows: maxTurnBytes === 2 ? 1 : 0 });
+  });
+}
+
 for (const ending of ["\n", ""] as const) {
   for (const fragmented of [false, true]) {
     test(`Command Code admits an exact-limit ${fragmented ? "fragmented" : "whole"} record with ${JSON.stringify(ending)} termination`, async () => {