fix(command-code): avoid double charging live NDJSON buffers

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",

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();

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

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",

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<Uint8Array>({
+    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);
+});