From da783b4a1993abd7525f8a6bece2cd56c8b3bdc4 Mon Sep 17 00:00:00 2001 From: StreamClash Dev Date: Wed, 21 Oct 2026 16:45:36 -0400 Subject: [PATCH 1/4] perf(chat): avoid rescanning fragmented SSE carry Co-authored-by: Platinum-Scripts <133908028+Platinum-Scripts@users.noreply.github.com> --- src/adapters/openai-chat.ts | 10 ++- structure/providers/mimo-bare-calls.md | 1 + structure/transports/byte-accounting.md | 10 +++ .../openai-chat-parallel-stream.test.ts | 76 ++++++++++++++++++- 4 files changed, 93 insertions(+), 4 deletions(-) diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index 9e794710c57..ba77d195cac 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -614,9 +614,13 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd } bufferBytes = nextBufferBytes; - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; - const residualBytes = budgetEncoder.encode(buffer).byteLength; + // The carry contains no newline after the previous drain. Without a new + // newline, keep its already-counted bytes instead of splitting and encoding + // the growing unfinished line again. TextDecoder emits complete code points. + const hasLineBreak = decoded.includes("\n"); + const lines = hasLineBreak ? buffer.split("\n") : []; + if (hasLineBreak) buffer = lines.pop() ?? ""; + const residualBytes = hasLineBreak ? budgetEncoder.encode(buffer).byteLength : bufferBytes; const residualReservation = budget.reserveTransient(residualBytes, { kind: "live_transient" }); residualReservation.commitRetained(); budget.releaseRetained(bufferBytes, { kind: "live_transient" }); diff --git a/structure/providers/mimo-bare-calls.md b/structure/providers/mimo-bare-calls.md index 14ce8b1654c..fbe957db0cf 100644 --- a/structure/providers/mimo-bare-calls.md +++ b/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 `` blocks must name an exact declared freeform wire tool. They may follow prose on the same line; diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 7111da2a96f..cd24dc9a2b2 100644 --- a/structure/transports/byte-accounting.md +++ b/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. diff --git a/tests/adapters/openai/openai-chat-parallel-stream.test.ts b/tests/adapters/openai/openai-chat-parallel-stream.test.ts index 9e47b3a7425..7bc30b153f8 100644 --- a/tests/adapters/openai/openai-chat-parallel-stream.test.ts +++ b/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({ 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({ 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({ 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); + } +}); From 03bb661b85c32c4953dcba5aad833213e247fcc6 Mon Sep 17 00:00:00 2001 From: StreamClash Dev Date: Wed, 21 Oct 2026 16:55:33 -0400 Subject: [PATCH 2/4] perf(chat): retain carry fast path within file-size cap Co-authored-by: Platinum-Scripts <133908028+Platinum-Scripts@users.noreply.github.com> --- src/adapters/openai-chat.ts | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index ba77d195cac..99663eee60e 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -614,13 +614,9 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd } bufferBytes = nextBufferBytes; - // The carry contains no newline after the previous drain. Without a new - // newline, keep its already-counted bytes instead of splitting and encoding - // the growing unfinished line again. TextDecoder emits complete code points. - const hasLineBreak = decoded.includes("\n"); - const lines = hasLineBreak ? buffer.split("\n") : []; - if (hasLineBreak) buffer = lines.pop() ?? ""; - const residualBytes = hasLineBreak ? budgetEncoder.encode(buffer).byteLength : bufferBytes; + const lines = decoded.includes("\n") ? buffer.split("\n") : [buffer]; + buffer = lines.pop() ?? ""; + 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" }); From 9f76f5d7cb1c0f727753185f569f3cf57523b0bf Mon Sep 17 00:00:00 2001 From: StreamClash Dev Date: Wed, 21 Oct 2026 17:15:42 -0400 Subject: [PATCH 3/4] perf(sse): avoid flattening fragmented block carry Co-authored-by: Platinum-Scripts <133908028+Platinum-Scripts@users.noreply.github.com> --- src/server/sse-payload-rewrite.ts | 15 +++- structure/transports/byte-accounting.md | 5 ++ tests/responses/sse-payload-rewrite.test.ts | 79 +++++++++++++++++++++ 3 files changed, 98 insertions(+), 1 deletion(-) diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index 99c67056eaa..d476b7e16cc 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/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; @@ -125,17 +127,24 @@ export function createSseBlockBuffer( reservation.commitRetained(); budget.releaseRetained(bufferBytes, scope); bufferBytes = nextBytes; + lastCodeUnit = fragment.charCodeAt(fragment.length - 1); + needsScan ||= fragment.includes("\r") || fragment.includes("\n"); } catch (error) { reservation.release(); throw error; } }, 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; }, }; diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index cd24dc9a2b2..a93ac948346 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -148,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. diff --git a/tests/responses/sse-payload-rewrite.test.ts b/tests/responses/sse-payload-rewrite.test.ts index 92eef790e78..0b2bd55a17f 100644 --- a/tests/responses/sse-payload-rewrite.test.ts +++ b/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 }); From daf0c1d897cd66bf3dd4224225c813da5c5fad3f Mon Sep 17 00:00:00 2001 From: JUN Date: Fri, 23 Oct 2026 03:51:17 +0900 Subject: [PATCH 4/4] fix(sse): keep block-buffer boundary state in step with the carry Record the cached final code unit and scan eligibility as soon as the fragment joins the buffer, before the budget commit, so a throwing commit cannot leave them describing an older carry. Co-authored-by: Platinum-Scripts <133908028+Platinum-Scripts@users.noreply.github.com> --- src/server/sse-payload-rewrite.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/server/sse-payload-rewrite.ts b/src/server/sse-payload-rewrite.ts index d476b7e16cc..8bb15a0302e 100644 --- a/src/server/sse-payload-rewrite.ts +++ b/src/server/sse-payload-rewrite.ts @@ -124,11 +124,11 @@ 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; - lastCodeUnit = fragment.charCodeAt(fragment.length - 1); - needsScan ||= fragment.includes("\r") || fragment.includes("\n"); } catch (error) { reservation.release(); throw error;