From c1b809270011a70fc58a4f058e59cdcd2440098f Mon Sep 17 00:00:00 2001 From: 0disoft Date: Sat, 7 Nov 2026 19:04:04 +0900 Subject: [PATCH] perf(command-code): measure fragmented NDJSON incrementally --- src/adapters/command-code.ts | 13 +++-- structure/transports/byte-accounting.md | 5 +- .../command-code-buffer-budget.test.ts | 57 ++++++++++++++++++- 3 files changed, 68 insertions(+), 7 deletions(-) diff --git a/src/adapters/command-code.ts b/src/adapters/command-code.ts index e116645cc50..e01cf0f8d53 100644 --- a/src/adapters/command-code.ts +++ b/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(); diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 4eb5fef7a5e..7037df8712e 100644 --- a/structure/transports/byte-accounting.md +++ b/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 diff --git a/tests/providers/command-code-buffer-budget.test.ts b/tests/providers/command-code-buffer-budget.test.ts index a7632986f0d..4f26461871e 100644 --- a/tests/providers/command-code-buffer-budget.test.ts +++ b/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 () => {