fix(stream): parse the trailing NDJSON line an Ollama stream leaves behind

createSSEStream splits on "\n" and keeps the remainder, which only flush()
parses. That call omitted targetFormat, so parseSSELine required a "data: "
prefix and dropped whatever an NDJSON provider left without a closing
newline. The !parsed.done guard compounded it: the SSE sentinel and an
Ollama final chunk both carry done:true, but the latter is the real last
chunk holding done_reason and the token counts.

Pass targetFormat and scope the sentinel check to formats that emit one, so
the tail reaches the translator. Accumulate its usage into state the same way
the transform loop does, so finalizeStream logs those tokens instead of null.

open-sse/utils/stream.js

@@ -408,8 +408,21 @@ export function createSSEStream(options = {}) {
         }
 
         if (buffer.trim()) {
-          const parsed = parseSSELine(buffer.trim());
-          if (parsed && !parsed.done) {
+          // Same parse as the transform loop: without targetFormat this only
+          // accepts "data: " lines, so an NDJSON provider (Ollama) lost whatever
+          // arrived without its closing newline.
+          const parsed = parseSSELine(buffer.trim(), targetFormat);
+          // parseSSELine turns the SSE sentinel "data: [DONE]" into { done: true },
+          // which must not be translated. An Ollama chunk also carries done:true,
+          // but it is the real final chunk — it holds finish_reason and the token
+          // counts — so it has to go through.
+          const isDoneSentinel = parsed?.done && targetFormat !== FORMATS.OLLAMA;
+          if (parsed && !isDoneSentinel) {
+            // Same accumulation the transform loop does, so finalizeStream() can
+            // log a tail chunk's tokens instead of falling back to null.
+            const extracted = extractUsage(parsed);
+            if (extracted) state.usage = mergeUsage(state.usage, extracted);
+
             const translated = translateResponse(targetFormat, sourceFormat, parsed, state);
 
             if (translated?._openaiIntermediate) {

tests/unit/ollama-stream-tail.test.js

@@ -0,0 +1,96 @@
+import { describe, expect, it } from "vitest";
+
+import { FORMATS } from "../../open-sse/translator/formats.js";
+import { createSSETransformStreamWithLogger } from "../../open-sse/utils/stream.js";
+
+// Ollama streams NDJSON — one raw JSON object per line, no "data: " prefix.
+// Whatever arrives without a closing newline stays in the line buffer and is
+// only parsed when the transform flushes.
+async function runOllamaStream(input) {
+  const encoder = new TextEncoder();
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.enqueue(encoder.encode(input));
+      controller.close();
+    },
+  });
+
+  const output = stream.pipeThrough(
+    createSSETransformStreamWithLogger(FORMATS.OLLAMA, FORMATS.OPENAI, "ollama", null, null, "gpt-oss:120b"),
+  );
+
+  const reader = output.getReader();
+  const decoder = new TextDecoder();
+  let text = "";
+  for (;;) {
+    const { value, done } = await reader.read();
+    if (done) break;
+    text += decoder.decode(value, { stream: true });
+  }
+  return text + decoder.decode();
+}
+
+const chunk = (content, done = false) => JSON.stringify({
+  model: "gpt-oss:120b",
+  created_at: "2026-08-25T00:00:00Z",
+  message: { role: "assistant", content },
+  done,
+  ...(done ? { done_reason: "stop", prompt_eval_count: 11, eval_count: 7 } : {}),
+});
+
+const deltas = (sse) => sse
+  .split("\n")
+  .filter((l) => l.startsWith("data: ") && l !== "data: [DONE]")
+  .map((l) => JSON.parse(l.slice(6)));
+
+describe("Ollama NDJSON stream: the tail left in the line buffer", () => {
+  it("delivers a content chunk that arrived without its newline", async () => {
+    const out = await runOllamaStream([chunk("hello"), chunk(" world")].join("\n"));
+    const content = deltas(out).map((c) => c.choices?.[0]?.delta?.content || "").join("");
+    expect(content).toBe("hello world");
+  });
+
+  it("delivers the final chunk — finish_reason and usage — when it arrives without its newline", async () => {
+    const out = await runOllamaStream([chunk("hello"), chunk("", true)].join("\n"));
+    const last = deltas(out).at(-1);
+    expect(last.choices[0].finish_reason).toBe("stop");
+    expect(last.usage).toEqual({ prompt_tokens: 11, completion_tokens: 7, total_tokens: 18 });
+  });
+
+  it("is unchanged when every line is newline-terminated", async () => {
+    const out = await runOllamaStream(`${[chunk("hello"), chunk(" world"), chunk("", true)].join("\n")}\n`);
+    const parsed = deltas(out);
+    expect(parsed.map((c) => c.choices?.[0]?.delta?.content || "").join("")).toBe("hello world");
+    expect(parsed.at(-1).choices[0].finish_reason).toBe("stop");
+    expect(parsed.at(-1).usage.total_tokens).toBe(18);
+  });
+});
+
+describe("SSE providers keep their sentinel handling", () => {
+  it("does not translate a trailing data: [DONE]", async () => {
+    const encoder = new TextEncoder();
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(encoder.encode(
+          `data: ${JSON.stringify({ choices: [{ delta: { content: "hi" } }] })}\ndata: [DONE]`,
+        ));
+        controller.close();
+      },
+    });
+    const out = stream.pipeThrough(
+      createSSETransformStreamWithLogger(FORMATS.OPENAI, FORMATS.OPENAI, "openai", null, null, "gpt-4o"),
+    );
+    const reader = out.getReader();
+    const decoder = new TextDecoder();
+    let text = "";
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      text += decoder.decode(value, { stream: true });
+    }
+    text += decoder.decode();
+    expect(text).toContain('"content":"hi"');
+    // The sentinel is a framing marker, not a chunk — it must not be translated.
+    expect(text).not.toContain('"done":true');
+  });
+});