From a1d6e6a53bc60b64e81875922dae53857e9ee0b0 Mon Sep 17 00:00:00 2001 From: 0disoft Date: Sat, 31 Oct 2026 17:10:59 +0900 Subject: [PATCH 1/2] fix(command-code): reserve only live buffer growth --- scripts/test-layout/layout.json | 2 +- src/adapters/command-code.ts | 20 ++--- structure/transports/byte-accounting.md | 8 ++ tests/fixtures/test-layout-expected.json | 2 +- .../command-code-buffer-budget.test.ts | 79 +++++++++++++++++++ 5 files changed, 100 insertions(+), 11 deletions(-) create mode 100644 tests/providers/command-code-buffer-budget.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 3232f870b0f..43af804cce6 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 66293ec5ef1..86c1cd11637 100644 --- a/src/adapters/command-code.ts +++ b/src/adapters/command-code.ts @@ -424,27 +424,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 ff55305299f..b0dc3e6122f 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -140,6 +140,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. + `src/lib/pinned-http.ts` pauses its socket response when the receiving Web Stream queue fills and resumes when the reader asks for more. For gzip and deflate, the final decoded queue also gates subsequent socket input even when the decoder would accept more coded bytes. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index b97dd7e3e9c..ba6682a4642 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -722,7 +722,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..eec46ee20dd --- /dev/null +++ b/tests/providers/command-code-buffer-budget.test.ts @@ -0,0 +1,79 @@ +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) { + let offset = 0; + let cancelled = false; + const body = new ReadableStream({ + pull(controller) { + 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); + }); +} + +test("Command Code buffer replacement preserves another owner's retained bytes", async () => { + const bytes = encoder.encode(line); + const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length + 64 }); + budget.chargeRetained(64, { kind: "retained_collectors" }); + for await (const _event of adapter().parseStream(input(bytes, 1).response, budget)) { /* drain */ } + expect(budget.snapshot()).toMatchObject({ currentBytes: 64, highWaterBytes: bytes.length + 64, overflows: 0 }); + budget.releaseRetained(64, { kind: "retained_collectors" }); + expect(budget.snapshot().currentBytes).toBe(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); +}); From ff0004da37ac8e2ecb6242c11ab8285816786929 Mon Sep 17 00:00:00 2001 From: 0disoft Date: Sat, 31 Oct 2026 17:37:30 +0900 Subject: [PATCH 2/2] test(command-code): cover shared buffer ownership and drain --- .../command-code-buffer-budget.test.ts | 49 +++++++++++++++---- 1 file changed, 40 insertions(+), 9 deletions(-) diff --git a/tests/providers/command-code-buffer-budget.test.ts b/tests/providers/command-code-buffer-budget.test.ts index eec46ee20dd..a7632986f0d 100644 --- a/tests/providers/command-code-buffer-budget.test.ts +++ b/tests/providers/command-code-buffer-budget.test.ts @@ -8,11 +8,12 @@ const content = "é한😀".repeat(32); const line = JSON.stringify({ type: "text-delta", text: content }); const encoder = new TextEncoder(); -function input(bytes: Uint8Array, chunkSize: number) { +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)); @@ -56,14 +57,44 @@ for (const fragmented of [false, true]) { }); } -test("Command Code buffer replacement preserves another owner's retained bytes", async () => { - const bytes = encoder.encode(line); - const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length + 64 }); - budget.chargeRetained(64, { kind: "retained_collectors" }); - for await (const _event of adapter().parseStream(input(bytes, 1).response, budget)) { /* drain */ } - expect(budget.snapshot()).toMatchObject({ currentBytes: 64, highWaterBytes: bytes.length + 64, overflows: 0 }); - budget.releaseRetained(64, { kind: "retained_collectors" }); - expect(budget.snapshot().currentBytes).toBe(0); +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 () => {