perf(sse): avoid rescanning and flattening fragmented SSE carry (carry #7497, #7498)

src/adapters/openai-chat.ts

@@ -614,9 +614,9 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd
           }
           bufferBytes = nextBufferBytes;
 
-          const lines = buffer.split("\n");
+          const lines = decoded.includes("\n") ? buffer.split("\n") : [buffer];
           buffer = lines.pop() ?? "";
-          const residualBytes = budgetEncoder.encode(buffer).byteLength;
+          const residualBytes = lines.length ? budgetEncoder.encode(buffer).byteLength : bufferBytes;
           const residualReservation = budget.reserveTransient(residualBytes, { kind: "live_transient" });
           residualReservation.commitRetained();
           budget.releaseRetained(bufferBytes, { kind: "live_transient" });

src/server/sse-payload-rewrite.ts

@@ -96,6 +96,8 @@ export function createSseBlockBuffer(
   let buffer = "";
   let offset = 0;
   let scanOffset = 0;
+  let needsScan = false;
+  let lastCodeUnit = Number.NaN;
   let bufferBytes = 0;
   const delimiterPattern = /(?:\r\n|\r(?!\n)|\n){2}/g;
 
@@ -111,7 +113,7 @@ export function createSseBlockBuffer(
       if (!fragment) return;
       let fragmentBytes = Buffer.byteLength(fragment, "utf8");
       // Decoder output never splits a surrogate pair, but keep the helper exact for string callers.
-      const last = buffer.charCodeAt(buffer.length - 1);
+      const last = lastCodeUnit;
       const first = fragment.charCodeAt(0);
       if (offset < buffer.length && last >= 0xd800 && last <= 0xdbff && first >= 0xdc00 && first <= 0xdfff) {
         fragmentBytes -= 2;
@@ -122,6 +124,8 @@ export function createSseBlockBuffer(
       try {
         compact();
         buffer += fragment;
+        lastCodeUnit = fragment.charCodeAt(fragment.length - 1);
+        needsScan ||= fragment.includes("\r") || fragment.includes("\n");
         reservation.commitRetained();
         budget.releaseRetained(bufferBytes, scope);
         bufferBytes = nextBytes;
@@ -131,11 +135,16 @@ export function createSseBlockBuffer(
       }
     },
     next() {
+      if (!needsScan) {
+        scanOffset = Math.max(offset, buffer.length - 3);
+        return null;
+      }
       delimiterPattern.lastIndex = scanOffset;
       const match = delimiterPattern.exec(buffer);
       if (!match) {
         // A delimiter is at most four code units. Revisit only its possible split prefix.
         scanOffset = Math.max(offset, buffer.length - 3);
+        needsScan = false;
         return null;
       }
       const start = match.index;
@@ -153,6 +162,8 @@ export function createSseBlockBuffer(
         buffer = "";
         offset = 0;
         scanOffset = 0;
+        needsScan = false;
+        lastCodeUnit = Number.NaN;
       }
       return { block, delimiter };
     },
@@ -164,6 +175,8 @@ export function createSseBlockBuffer(
       buffer = "";
       offset = 0;
       scanOffset = 0;
+      needsScan = false;
+      lastCodeUnit = Number.NaN;
       bufferBytes = 0;
     },
   };

structure/providers/mimo-bare-calls.md

@@ -5,6 +5,7 @@
 Standalone recovery (#7260) is separate from
 [structured-call/identified-wrapper reconciliation](chat-compat.md#serialized-tool-call-content).
 `src/adapters/openai-chat.ts` composes both paths without changing the identified-wrapper contract.
+Its fragmented-line carry follows the [stream-buffer accounting contract](../transports/byte-accounting.md#stream-buffer-accounting); avoiding repeated carry scans does not change reconciliation or terminal dispatch.
 `src/adapters/openai-chat/standalone-bare-tool-calls.ts` enables it only for MiMo `mimo-v2` or
 dotted `mimo-v2.*` model IDs when no structured calls exist. Complete bare `<tool_call>` blocks
 must name an exact declared freeform wire tool. They may follow prose on the same line;

structure/transports/byte-accounting.md

@@ -121,6 +121,16 @@ lifecycle, cancellation races, protocol envelopes, and the real HTTP admission b
 
 ## Stream-buffer accounting
 
+The Chat adapter's decoded carry contains only an unfinished line after each drain.
+A network fragment with no newline retains the carry's already-counted UTF-8 size
+instead of splitting and re-encoding the entire growing line. The append and
+residual overlap reservations remain in their original order, including the
+otherwise redundant reservation, so existing cap failures and high-water accounting
+are unchanged. A fragment containing a newline drains through the existing parser.
+Unicode decoder boundaries, CRLF handling, EOF, terminal repairs and call ordering
+retain their contracts. `tests/adapters/openai/openai-chat-parallel-stream.test.ts`
+covers fragmented work scaling and the Unicode/EOF boundary cases.
+
 HTTP owners in `src/server/responses/core.ts`, `src/server/chat-completions.ts` and `src/server/claude-messages.ts` pass the request abort signal to `src/lib/translator-budget.ts`. An aborted request releases its owned translator budget even when its response body is never consumed. Abort finalization changes accounting only; the transport keeps ownership of producer cancellation and response terminal precedence. Explicit body cancellation also releases the budget before awaiting upstream reader cancellation. EOF and read errors share the same idempotent finalizer, which removes its abort listener. Responses lifetime wrapping in `src/server/responses/core-lifetime.ts` preserves native, eager-relay, preinspected-response, run-turn and nonreplayable markers, including through an outer serving-account lease. Both budget and account-lease finalization also carry replay-refusal provenance with `carryReplayRefusal`: deferred logging distinguishes synthetic transport-ambiguous refusals from genuine upstream 429 rate limits without changing the client response or replay policy. `tests/responses/responses-translated-reset.test.ts` covers attribution through each wrapper and their composition. `tests/adapters/translator-budget-lifetime.test.ts` holds cancellation pending and covers abort before wrapping, abort after wrapping, explicit cancellation, marker preservation and successful EOF.
 
 Disposed budgets ignore late call, reservation, charge and observation work; reservations created before disposal cannot resurrect per-turn or aggregate counters. The same Bun contract checks both accounting scopes before any producer cancellation and while an explicit cancellation is pending. Every producer must honor its transport abort link; disposed accounting does not provide a byte cap to work that outlives its request.
@@ -138,6 +148,11 @@ cancellation or consumer closure; a buffer-limit failure terminates without anot
 `src/server/sse-payload-rewrite.ts` shares an incremental block buffer with native Chat. It scans
 only new input plus a three-character delimiter prefix and counts consumed blocks rather than
 remaining suffixes. CR, LF and CRLF share partial-event, injection/drop and EOF behavior.
+After an unsuccessful delimiter search, fragments without CR or LF cannot create a delimiter;
+the buffer advances its scan cursor without passing the growing carry to the regex again.
+Queued appends retain scan eligibility until drained. This adds no payload buffer or budget operation.
+Surrogate-boundary accounting caches the final code unit from each fragment rather than indexing
+the growing carry. A failed scan retains that boundary; emptying or clearing the buffer resets it.
 Output admission precedes its single UTF-8 encoding;
 failed enqueue and cancellation release the reservation without re-entering a disposed rewrite.
 Old/new buffer overlap remains charged against the same translator cap.

tests/adapters/openai/openai-chat-parallel-stream.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, test } from "bun:test";
+import { describe, expect, spyOn, test } from "bun:test";
 import { createOpenAIChatAdapter as createOpenAIChatAdapterProduction } from "../../../src/adapters/openai-chat";
 import type { TranslatorBudget } from "../../../src/lib/translator-budget";
 import type { AdapterEvent } from "../../../src/types";
@@ -460,3 +460,77 @@ data: [DONE]
     expect(budget.snapshot()).toMatchObject({ activeCalls: 0, currentBytes: 0, overflows: limit === 9 ? 1 : 0 });
   });
 });
+
+
+test("fragmented large SSE lines bound UTF-8 counting work without changing tool arguments", async () => {
+  const args = JSON.stringify({ code: "x".repeat(256 * 1024) });
+  const wire = new TextEncoder().encode(sse([
+    chunkOf([{ index: 0, id: "call_fragmented", function: { name: "review", arguments: args } }]),
+    chunkOf([], "tool_calls"),
+  ]));
+  let cursor = 0;
+  const source = new ReadableStream<Uint8Array>({ pull(controller) {
+    if (cursor === wire.length) { controller.close(); return; }
+    const next = Math.min(cursor + 512, wire.length);
+    controller.enqueue(wire.subarray(cursor, next)); cursor = next;
+  } });
+  let encodedCharacters = 0;
+  const original = TextEncoder.prototype.encode;
+  const counting = spyOn(TextEncoder.prototype, "encode").mockImplementation(function (this: TextEncoder, input?: string) {
+    encodedCharacters += input?.length ?? 0;
+    return original.call(this, input);
+  });
+  try {
+    const events: AdapterEvent[] = [];
+    for await (const event of createOpenAIChatAdapter(provider).parseStream(new Response(source))) events.push(event);
+    expect(assembled(events)).toEqual([{ id: "call_fragmented", name: "review", args }]);
+    expect(events.at(-1)?.type).toBe("done");
+    // A deterministic work bound, not a wall-clock test. Re-encoding the growing
+    // unfinished line for every 512-byte fragment exceeds this by orders of magnitude.
+    expect(encodedCharacters).toBeLessThan(wire.length * 8);
+  } finally { counting.mockRestore(); }
+});
+
+
+test("fragmented SSE carry preserves Unicode, CRLF and unterminated EOF frames", async () => {
+  for (const size of [1, 2, 3, 7, 512]) {
+    const args = JSON.stringify({ code: "é🐢汉字\r\n" });
+    const wire = new TextEncoder().encode(sse([
+      chunkOf([{ index: 0, id: "unicode", function: { name: "review", arguments: args } }]),
+      chunkOf([], "tool_calls"),
+    ]).replaceAll("\n", "\r\n").trimEnd());
+    let cursor = 0;
+    const stream = new ReadableStream<Uint8Array>({ pull(controller) {
+      if (cursor >= wire.length) { controller.close(); return; }
+      controller.enqueue(wire.subarray(cursor, cursor + size)); cursor += size;
+    } });
+    const events: AdapterEvent[] = [];
+    const budget = createTestTranslatorBudget();
+    for await (const event of createOpenAIChatAdapter(provider).parseStream(new Response(stream), budget)) events.push(event);
+    expect(assembled(events)).toEqual([{ id: "unicode", name: "review", args }]);
+    expect(events.at(-1)?.type).toBe("done");
+    expect(budget.snapshot().currentBytes).toBe(0);
+  }
+});
+
+
+test("fragmented invalid UTF-8 keeps decoder replacement and carry accounting", async () => {
+  const wire = new TextEncoder().encode(sse([
+    chunkOf([{ index: 0, id: "replacement", function: { name: "review", arguments: JSON.stringify({ code: "MARK" }) } }]),
+    chunkOf([], "tool_calls"),
+  ]));
+  const at = new TextDecoder().decode(wire).indexOf("MARK");
+  wire.set([0xed, 0xa0, 0x80], at); // Invalid UTF-8 surrogate, not three ASCII bytes.
+  const expected = JSON.stringify({ code: "���K" });
+  for (const size of [1, 2, 7]) {
+    let cursor = 0;
+    const stream = new ReadableStream<Uint8Array>({ pull(controller) {
+      if (cursor >= wire.length) { controller.close(); return; }
+      controller.enqueue(wire.subarray(cursor, cursor + size)); cursor += size;
+    } });
+    const events: AdapterEvent[] = [];const budget = createTestTranslatorBudget();
+    for await (const event of createOpenAIChatAdapter(provider).parseStream(new Response(stream), budget)) events.push(event);
+    expect(assembled(events)).toEqual([{ id: "replacement", name: "review", args: expected }]);
+    expect(events.at(-1)?.type).toBe("done");expect(budget.snapshot().currentBytes).toBe(0);
+  }
+});

tests/responses/sse-payload-rewrite.test.ts

@@ -176,6 +176,85 @@ describe("SSE payload rewrite composition", () => {
     }
   });
 
+  test("does not send a growing unfinished frame to the delimiter regex on each fragment", () => {
+    const text = `data: ${"x".repeat(256 * 1024)}\r\n\r\n`;
+    const budget = createTestTranslatorBudget();
+    const buffer = createSseBlockBuffer(budget);
+    const original = RegExp.prototype.exec;
+    const pattern = /(?:\r\n|\r(?!\n)|\n){2}/g.source;
+    let regexInputUnits = 0;
+    const exec = spyOn(RegExp.prototype, "exec").mockImplementation(function (this: RegExp, input: string) {
+      if (this.source === pattern && this.global) regexInputUnits += input.length;
+      return original.call(this, input);
+    });
+    try {
+      let delivered = "";
+      for (let index = 0; index < text.length; index += 512) {
+        buffer.append(text.slice(index, index + 512));
+        let block;
+        while ((block = buffer.next())) delivered += block.block + block.delimiter;
+      }
+      expect(delivered).toBe(text);
+      // Deterministic work bound: even a near-end regex search can flatten a
+      // growing rope, so count the full subject passed to that native operation.
+      expect(regexInputUnits).toBeLessThan(text.length * 4);
+      expect(budget.snapshot().currentBytes).toBe(0);
+    } finally { exec.mockRestore(); buffer.clear(); }
+  });
+
+  test("append boundary checks do not index into the growing carry on each fragment", () => {
+    const text = `data: ${"x".repeat(256 * 1024)}\n\n`;
+    const budget = createTestTranslatorBudget(), buffer = createSseBlockBuffer(budget);
+    const original = String.prototype.charCodeAt;
+    let indexedSubjectUnits = 0;
+    const charCodeAt = spyOn(String.prototype, "charCodeAt").mockImplementation(function (this: string, position: number) {
+      indexedSubjectUnits += this.length;
+      return original.call(this, position);
+    });
+    try {
+      let delivered = "";
+      for (let index = 0; index < text.length; index += 512) {
+        buffer.append(text.slice(index, index + 512));
+        let block;
+        while ((block = buffer.next())) delivered += block.block + block.delimiter;
+      }
+      expect(delivered).toBe(text);
+      expect(indexedSubjectUnits).toBeLessThan(text.length * 4);
+      expect(budget.snapshot().currentBytes).toBe(0);
+    } finally { charCodeAt.mockRestore(); buffer.clear(); }
+  });
+
+  test("a failed delimiter scan retains the surrogate boundary of the carry", () => {
+    const budget = createTestTranslatorBudget(), buffer = createSseBlockBuffer(budget);
+    buffer.append("data: \r\ud800");
+    expect(buffer.next()).toBeNull();
+    buffer.append("\udc00");
+    expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength("data: \r\ud800\udc00", "utf8"));
+    buffer.clear();
+    expect(budget.snapshot().currentBytes).toBe(0);
+  });
+
+  test("scan eligibility survives queued appends, compaction and clear", () => {
+    const budget = createTestTranslatorBudget();
+    const buffer = createSseBlockBuffer(budget);
+    let reference = "";
+    for (const fragments of [["data: one", "x", "\r", "\n", "\r", "\n", "tail"], ["plain", "x", "\n", "\n", "data: two\r\r", "last"], ["\ud800", "\udc00", "\n", "\n"]]) {
+      for (const fragment of fragments) { buffer.append(fragment); reference += fragment; }
+      for (;;) {
+        const expected = nextSseBlock(reference), actual = buffer.next();
+        expect(actual).toEqual(expected && {block: expected.block, delimiter: expected.delimiter});
+        if (!expected) break;
+        reference = expected.rest;
+        buffer.compact();
+      }
+      expect(buffer.tail()).toBe(reference);
+      expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(reference, "utf8"));
+      buffer.clear(); reference = "";
+      buffer.append("plain"); expect(buffer.next()).toBeNull(); buffer.clear();
+    }
+    expect(budget.snapshot().currentBytes).toBe(0);
+  });
+
   test("rejects an oversized replacement before encoding or delivering it", async () => {
     const upstream = streamFromText("data: small\n\n");
     const budget = createTestTranslatorBudget({ maxTurnBytes: 64 });