diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index c492598af67..67bb3d24d59 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -753,7 +753,7 @@ "combo-workspace-data.test.ts": "gui", "combo-workspace-jev-decision.test.ts": "gui", "combos.test.ts": "codex-integration", "combos-kiro-profile-hop.test.ts": "codex-integration", - "command-code-error-finish.test.ts": "providers", + "command-code-error-finish.test.ts": "providers", "command-code-buffer-budget.test.ts": "providers", "command-code-efforts.test.ts": "providers", "command-code-project-context.test.ts": "providers", "command-code-provider.test.ts": "providers", diff --git a/src/adapters/command-code.ts b/src/adapters/command-code.ts index 8a26c300832..e116645cc50 100644 --- a/src/adapters/command-code.ts +++ b/src/adapters/command-code.ts @@ -425,27 +425,29 @@ async function*ndjson(response: Response, budget: TranslatorBudget): AsyncGenera const encoder = new TextEncoder(); let buffer = ""; let bufferBytes = 0; + // One live buffer owns one charge; resizing must not reserve old + replacement together. + const resizeBuffer = (nextBytes: number) => { + if (nextBytes > bufferBytes) { + budget.reserveTransient(nextBytes - bufferBytes, { kind: "live_transient" }).commitRetained(); + } else { + budget.releaseRetained(bufferBytes - nextBytes, { kind: "live_transient" }); + } + bufferBytes = nextBytes; + }; try { for (;;) { const { value, done } = await reader.read(); const next = buffer + decoder.decode(value, { stream: !done }); const nextBytes = encoder.encode(next).byteLength; - const reservation = budget.reserveTransient(nextBytes, { kind: "live_transient" }); + resizeBuffer(nextBytes); buffer = next; - reservation.commitRetained(); - budget.releaseRetained(bufferBytes, { kind: "live_transient" }); - bufferBytes = nextBytes; let newline = buffer.indexOf("\n"); while (newline >= 0) { const line = buffer.slice(0, newline).trim(); buffer = buffer.slice(newline + 1); if (line) yield* decodeEventLine(line); newline = buffer.indexOf("\n"); } - const residualBytes = encoder.encode(buffer).byteLength; - const residualReservation = budget.reserveTransient(residualBytes, { kind: "live_transient" }); - residualReservation.commitRetained(); - budget.releaseRetained(bufferBytes, { kind: "live_transient" }); - bufferBytes = residualBytes; + 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 4da2a12db7d..a93cd509622 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -143,6 +143,14 @@ envelopes, and real HTTP streams; `tests/server/inbound-body-budget.test.ts` cov ## Stream-buffer accounting +Command Code NDJSON in `src/adapters/command-code.ts` holds one `live_transient` charge for +its decoded buffer. Growth reserves only the positive byte delta; draining releases only the +removed bytes. Recounting an unchanged residual does not reserve it twice. Failed growth keeps +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. + Native Ollama NDJSON in `src/adapters/ollama-native.ts` measures new decoded fragments without allocating UTF-8 arrays. A fragment without a newline extends the cached residual byte count; only a fragment containing a newline triggers splitting of the accumulated record. Complete diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 3854b21fc39..97bad7d34b2 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -741,7 +741,7 @@ "combo-vision-enrollment.test.ts": "routing", "combo-stream-preflight.test.ts": "routing", "combo-workspace-data.test.ts": "gui", "combo-workspace-jev-decision.test.ts": "gui", "combos.test.ts": "codex-integration", "combos-kiro-profile-hop.test.ts": "codex-integration", - "command-code-error-finish.test.ts": "providers", "command-code-credit-failover.test.ts": "providers", "command-code-credit-rotation.test.ts": "providers", "command-code-plan-denial.test.ts": "providers", + "command-code-error-finish.test.ts": "providers", "command-code-buffer-budget.test.ts": "providers", "command-code-credit-failover.test.ts": "providers", "command-code-credit-rotation.test.ts": "providers", "command-code-plan-denial.test.ts": "providers", "command-code-efforts.test.ts": "providers", "command-code-project-context.test.ts": "providers", "command-code-provider.test.ts": "providers", "command-code-quota.test.ts": "providers", "command-code-tool-text.test.ts": "providers", diff --git a/tests/providers/command-code-buffer-budget.test.ts b/tests/providers/command-code-buffer-budget.test.ts new file mode 100644 index 00000000000..a7632986f0d --- /dev/null +++ b/tests/providers/command-code-buffer-budget.test.ts @@ -0,0 +1,110 @@ +import { expect, test } from "bun:test"; +import { createCommandCodeAdapter } from "../../src/adapters/command-code"; +import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import type { AdapterEvent } from "../../src/types"; + +const adapter = () => createCommandCodeAdapter({ adapter: "command-code", baseUrl: "https://command.example" }); +const content = "é한😀".repeat(32); +const line = JSON.stringify({ type: "text-delta", text: content }); +const encoder = new TextEncoder(); + +function input(bytes: Uint8Array, chunkSize: number, beforeRead?: (offset: number) => void) { + let offset = 0; + let cancelled = false; + const body = new ReadableStream({ + pull(controller) { + beforeRead?.(offset); + if (offset === bytes.length) { controller.close(); return; } + const end = Math.min(bytes.length, offset + chunkSize); + controller.enqueue(bytes.subarray(offset, end)); + offset = end; + }, + cancel() { cancelled = true; }, + }, { highWaterMark: 0 }); + return { response: new Response(body), cancelled: () => cancelled }; +} + +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 () => { + const bytes = encoder.encode(line + ending); + const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length }); + const source = input(bytes, fragmented ? 1 : bytes.length); + const events: AdapterEvent[] = []; + for await (const event of adapter().parseStream(source.response, budget)) events.push(event); + expect(events).toEqual([ + { type: "text_delta", text: content }, + { type: "done", usage: undefined, stopReason: undefined }, + ]); + expect(budget.snapshot()).toMatchObject({ currentBytes: 0, highWaterBytes: bytes.length, overflows: 0 }); + }); + } +} + +for (const fragmented of [false, true]) { + test(`Command Code refuses a ${fragmented ? "fragmented" : "whole"} record one byte over budget and cleans up`, async () => { + const bytes = encoder.encode(line); + const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length - 1 }); + const source = input(bytes, fragmented ? 1 : bytes.length); + const events: AdapterEvent[] = []; + const consume = async () => { + for await (const event of adapter().parseStream(source.response, budget)) events.push(event); + }; + await expect(consume()).rejects.toMatchObject({ code: "translation_buffer_limit", kind: "live_transient" }); + expect(events).toHaveLength(0); + expect(budget.snapshot()).toMatchObject({ currentBytes: 0, overflows: 1 }); + expect(source.cancelled()).toBe(true); + }); +} + +for (const deficit of [0, 1]) { + test(`Command Code preserves another live_transient owner's bytes on ${deficit ? "refused" : "successful"} growth`, async () => { + const bytes = encoder.encode(line); + const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length + 64 - deficit }); + budget.chargeRetained(64, { kind: "live_transient" }); + const source = input(bytes, 1); + const events: AdapterEvent[] = []; + const consume = async () => { + for await (const event of adapter().parseStream(source.response, budget)) events.push(event); + }; + if (deficit) { + await expect(consume()).rejects.toMatchObject({ code: "translation_buffer_limit" }); + expect(events).toHaveLength(0); + expect(source.cancelled()).toBe(true); + } else { + await consume(); + expect(events.at(-1)?.type).toBe("done"); + } + expect(budget.snapshot()).toMatchObject({ currentBytes: 64, highWaterBytes: bytes.length + 64 - deficit, overflows: deficit }); + budget.releaseRetained(64, { kind: "live_transient" }); + expect(budget.snapshot().currentBytes).toBe(0); + }); +} + +test("Command Code releases drained bytes before reading the next fragment", async () => { + const carryBytes = 16; // The second record's ASCII JSON prefix, before any text content. + const firstReadBytes = encoder.encode(line + "\n").length + carryBytes; + const budget = createTestTranslatorBudget({ maxTurnBytes: firstReadBytes }); + const retainedBeforeSecondRead: number[] = []; + const source = input(encoder.encode(line + "\n" + line), firstReadBytes, offset => { + if (offset === firstReadBytes) retainedBeforeSecondRead.push(budget.snapshot().currentBytes); + }); + const events: AdapterEvent[] = []; + for await (const event of adapter().parseStream(source.response, budget)) events.push(event); + expect(retainedBeforeSecondRead).toEqual([carryBytes]); + expect(events.filter(event => event.type === "text_delta").map(event => event.text)).toEqual([content, content]); + expect(events.at(-1)?.type).toBe("done"); + expect(budget.snapshot()).toMatchObject({ currentBytes: 0, highWaterBytes: firstReadBytes, overflows: 0 }); +}); + +test("Command Code consumer return releases its retained record and cancels upstream", async () => { + const bytes = encoder.encode(line + "\n"); + const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length }); + const source = input(bytes, bytes.length); + const events = adapter().parseStream(source.response, budget); + expect((await events.next()).value).toMatchObject({ type: "text_delta", text: content }); + expect(budget.snapshot().currentBytes).toBe(bytes.length); + await events.return(); + expect(budget.snapshot().currentBytes).toBe(0); + expect(source.cancelled()).toBe(true); +});