perf(ollama): count fragmented NDJSON bytes incrementally

scripts/test-layout/layout.json

@@ -1331,7 +1331,7 @@
     "ocx-launcher-source.test.ts": "cli",
     "ocx-run.test.ts": "cli",
     "ollama-native-parser.test.ts": "providers/ollama", "ollama-native-reasoning-wire.test.ts": "providers/ollama", "ollama-native-structured-output.test.ts": "providers/ollama", "ollama-native-tool-continuations.test.ts": "providers/ollama",
-    "ollama-native-v4.test.ts": "providers/ollama", "ollama-native.test.ts": "providers/ollama", "ollama-show-enrichment-v7.test.ts": "providers/ollama", "ollama-show-enrichment.test.ts": "providers/ollama",
+    "ollama-native-v4.test.ts": "providers/ollama", "ollama-native-scaling.test.ts": "providers/ollama", "ollama-native.test.ts": "providers/ollama", "ollama-show-enrichment-v7.test.ts": "providers/ollama", "ollama-show-enrichment.test.ts": "providers/ollama",
     "ollama-show-ignore-abort.test.ts": "providers/ollama",
     "opaque-blob-wrapped-rejection.test.ts": "responses", "openai-api-virtual-models.test.ts": "adapters/openai", "openai-chat-bounded-tool-names.test.ts": "adapters/openai", "openai-chat-dangling-toolcalls.test.ts": "adapters/openai",
     "openai-chat-developer-position.test.ts": "adapters/openai", "openai-chat-eof.test.ts": "adapters/openai", "openai-chat-hardening.test.ts": "adapters/openai", "openai-empty-assistant-history.test.ts": "adapters/openai", "openai-chat-image-normalization.test.ts": "adapters/openai",

src/adapters/ollama-native.ts

@@ -1051,7 +1051,6 @@ async function* parseOllamaNativeStream(
   }
   const reader = response.body.getReader();
   const decoder = new TextDecoder("utf-8", { fatal: true });
-  const encoder = new TextEncoder();
   const state = streamState(allowParallelToolCalls);
   const issuedIds = issuedToolCallIds ?? new Set<string>();
   let buffer = "";
@@ -1060,15 +1059,20 @@ async function* parseOllamaNativeStream(
 
   const ingest = function* (text: string): Generator<AdapterEvent> {
     if (!text) return;
+    const hasNewline = text.includes("\n");
     buffer += text;
-    const lines = buffer.split("\n");
-    buffer = lines.pop() ?? "";
+    // After draining, the carry contains no newline. Only new text can complete a record.
+    const lines = hasNewline ? buffer.split("\n") : [];
+    if (hasNewline) buffer = lines.pop() ?? "";
     // The safety bound applies to the genuinely retained residual — the incomplete NDJSON record
     // still being assembled — and to each complete record below. It must never depend on the
     // transport read size (one read may carry many valid records) nor transiently charge
     // old + replacement together. On a ceiling violation the old reservation is left in place so
     // the generator's finally releases exactly what is held.
-    const residualBytes = encoder.encode(buffer).byteLength;
+    // TextDecoder emits complete code points; fragment byte counts add without a surrogate fixup.
+    const residualBytes = hasNewline
+      ? Buffer.byteLength(buffer, "utf8")
+      : bufferBytes + Buffer.byteLength(text, "utf8");
     if (residualBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) {
       throw new TranslatorBudgetExceededError("live_transient", TRANSLATOR_MAX_SSE_EVENT_BYTES);
     }
@@ -1077,7 +1081,7 @@ async function* parseOllamaNativeStream(
 
     for (const rawLine of lines) {
       const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
-      const lineBytes = encoder.encode(line).byteLength;
+      const lineBytes = Buffer.byteLength(line, "utf8");
       if (lineBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) {
         throw new TranslatorBudgetExceededError("live_transient", TRANSLATOR_MAX_SSE_EVENT_BYTES);
       }
@@ -1112,7 +1116,7 @@ async function* parseOllamaNativeStream(
       yield* ingest(decoder.decode());
     if (!state.terminal && buffer.length > 0) {
       const rawLine = buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer;
-      const lineBytes = encoder.encode(rawLine).byteLength;
+      const lineBytes = Buffer.byteLength(rawLine, "utf8");
       if (lineBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) throw new TranslatorBudgetExceededError("live_transient", TRANSLATOR_MAX_SSE_EVENT_BYTES);
       try {
         // The EOF record keeps its residual charge while it is parsed and consumed — releasing it

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
 
+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
+records and the remaining residual retain their independent size checks and existing reservation
+order, including CRLF and EOF handling. The fatal streaming decoder emits complete code points,
+so fragment counts remain exact across split UTF-8 reads. `tests/providers/ollama/ollama-native-scaling.test.ts`
+checks linear measurement and splitting work, framing equivalence, invalid UTF-8 and 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.

tests/fixtures/test-layout-expected.json

@@ -1071,6 +1071,7 @@
   "ocx-launcher-desktop-handoff.test.ts": "cli", "cli-path-diagnostics.test.ts": "cli",
   "ocx-run.test.ts": "cli", "ollama-native-parser.test.ts": "providers/ollama",
   "ollama-native-reasoning-wire.test.ts": "providers/ollama",
+  "ollama-native-scaling.test.ts": "providers/ollama",
   "ollama-native-structured-output.test.ts": "providers/ollama", "ollama-native-v4.test.ts": "providers/ollama",
   "ollama-native.test.ts": "providers/ollama", "ollama-show-enrichment-v7.test.ts": "providers/ollama",
   "ollama-show-enrichment.test.ts": "providers/ollama", "ollama-show-ignore-abort.test.ts": "providers/ollama",

tests/providers/ollama/ollama-native-scaling.test.ts

@@ -0,0 +1,131 @@
+import { expect, spyOn, test } from "bun:test";
+import { createOllamaNativeAdapter } from "../../../src/adapters/ollama-native";
+import type { AdapterEvent } from "../../../src/types";
+import { createTestTranslatorBudget } from "../../helpers/translator-budget";
+
+const encoder = new TextEncoder();
+const adapter = () => createOllamaNativeAdapter({ adapter: "ollama-native", baseUrl: "http://127.0.0.1:11434", models: ["fixture"] });
+const record = (content: string, done = true) => JSON.stringify({ model: "fixture", message: { role: "assistant", content }, done, ...(done ? { done_reason: "stop", eval_count: 7 } : {}) });
+
+function response(bytes: Uint8Array, chunkSize: number, emptyChunks = false, keepOpen = false) {
+  let offset = 0;
+  let empty = emptyChunks;
+  let cancelled = false;
+  const body = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (empty) { empty = false; controller.enqueue(new Uint8Array(0)); return; }
+      if (offset === bytes.length) { if (!keepOpen) controller.close(); return; }
+      const end = Math.min(bytes.length, offset + chunkSize);
+      controller.enqueue(bytes.subarray(offset, end));
+      offset = end;
+      empty = emptyChunks;
+    },
+    cancel() { cancelled = true; },
+  });
+  return { response: new Response(body), cancelled: () => cancelled };
+}
+
+async function collect(input: Response) {
+  const budget = createTestTranslatorBudget();
+  const events: AdapterEvent[] = [];
+  for await (const event of adapter().parseStream(input, budget)) events.push(event);
+  expect(budget.snapshot().currentBytes).toBe(0);
+  return events;
+}
+
+for (const metric of ["measurement", "splitting"] as const) {
+  test(`fragmented Ollama records have linear ${metric} work`, async () => {
+    const content = "x".repeat(256 * 1024);
+    const bytes = encoder.encode(record(content) + "\n");
+    const input = response(bytes, 256, false, true);
+    const encode = TextEncoder.prototype.encode;
+    const byteLength = Buffer.byteLength;
+    const split = String.prototype.split;
+    let measuredChars = 0;
+    let splitChars = 0;
+    const encoding = spyOn(TextEncoder.prototype, "encode").mockImplementation(function (value = "") {
+      measuredChars += value.length;
+      return Reflect.apply(encode, this, [value]);
+    });
+    const measuring = spyOn(Buffer, "byteLength").mockImplementation((value, encoding) => {
+      if (typeof value === "string") measuredChars += value.length;
+      return byteLength(value, encoding);
+    });
+    const splitting = spyOn(String.prototype, "split").mockImplementation(function (...args: Parameters<typeof split>) {
+      if (args[0] === "\n") splitChars += this.length;
+      return Reflect.apply(split, this, args);
+    });
+    let events: AdapterEvent[];
+    try { events = await collect(input.response); }
+    finally { encoding.mockRestore(); measuring.mockRestore(); splitting.mockRestore(); }
+    expect(events.filter(event => event.type === "text_delta").map(event => event.text).join("")).toBe(content);
+    expect(events.at(-1)).toMatchObject({ type: "done", usage: { outputTokens: 7 } });
+    expect(input.cancelled()).toBe(true);
+    // Bound actual work, independent of CPU speed. Both UTF-8 counting APIs are observed.
+    if (metric === "measurement") expect(measuredChars).toBeLessThanOrEqual(bytes.length * 8);
+    else expect(splitChars).toBeLessThanOrEqual(bytes.length * 2);
+  });
+}
+
+for (const ending of ["\n", "\r\n", ""] as const) {
+  test(`Ollama fragmented UTF-8 preserves BOM, blank lines and ${JSON.stringify(ending)} termination`, async () => {
+    const bytes = encoder.encode("\ufeff\n" + record("한글 😀", false) + "\r\n\n" + record("끝") + ending);
+    const whole = await collect(response(bytes, bytes.length).response);
+    const fragmented = await collect(response(bytes, 1, true).response);
+    expect(fragmented).toEqual(whole);
+    expect(fragmented.filter(event => event.type === "text_delta").map(event => event.text).join("")).toBe("한글 😀끝");
+    expect(fragmented.at(-1)?.type).toBe("done");
+  });
+}
+
+for (const deficit of [0, 1]) {
+  test(`fragmented Unicode residual ${deficit ? "rejects one byte below" : "admits at"} its UTF-8 byte limit`, async () => {
+    const content = "é한😀";
+    const text = record(content);
+    const bytes = encoder.encode(text);
+    expect(bytes.length).toBeGreaterThan(text.length);
+    const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length - deficit });
+    let offset = 0;
+    let cancelled = false;
+    let beforeEof: ReturnType<typeof budget.snapshot> | undefined;
+    const body = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (offset < bytes.length) {
+          controller.enqueue(bytes.subarray(offset, ++offset));
+          return;
+        }
+        // With no read-ahead, every fragment has been ingested; no delimiter or EOF yet.
+        beforeEof = budget.snapshot();
+        controller.close();
+      },
+      cancel() { cancelled = true; },
+    }, { highWaterMark: 0 });
+    const events: AdapterEvent[] = [];
+    for await (const event of adapter().parseStream(new Response(body), budget)) events.push(event);
+    if (deficit === 0) {
+      expect(beforeEof).toMatchObject({ currentBytes: bytes.length, highWaterBytes: bytes.length, overflows: 0 });
+      expect(events.filter(event => event.type === "text_delta").map(event => event.text).join("")).toBe(content);
+      expect(events.at(-1)?.type).toBe("done");
+      expect(budget.snapshot().highWaterBytes).toBe(bytes.length);
+    } else {
+      expect(beforeEof).toBeUndefined(); // Refusal must happen while the record is still buffered.
+      expect(events).toHaveLength(1);
+      expect(events[0]).toMatchObject({ type: "error", code: "translation_buffer_limit" });
+      expect(budget.snapshot().overflows).toBe(1);
+      expect(cancelled).toBe(true);
+    }
+    expect(budget.snapshot().currentBytes).toBe(0);
+  });
+}
+
+test("invalid UTF-8 after a retained partial record releases its budget and cancels", async () => {
+  const prefix = encoder.encode('{"message":{"content":"partial ');
+  const bytes = new Uint8Array(prefix.length + 1);
+  bytes.set(prefix);
+  bytes[prefix.length] = 0xff;
+  const input = response(bytes, prefix.length, false, true);
+  const events = await collect(input.response);
+  expect(events).toHaveLength(1);
+  expect(events[0]).toMatchObject({ type: "error" });
+  expect(input.cancelled()).toBe(true);
+});