From f9d82c65755a65329ca7a6632cbbdc79e00715c1 Mon Sep 17 00:00:00 2001 From: Nguyen Thanh Dat Date: Tue, 25 Aug 2026 13:35:48 +0700 Subject: [PATCH] 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 | 17 ++++- tests/unit/ollama-stream-tail.test.js | 96 +++++++++++++++++++++++++++ 2 files changed, 111 insertions(+), 2 deletions(-) create mode 100644 tests/unit/ollama-stream-tail.test.js diff --git a/open-sse/utils/stream.js b/open-sse/utils/stream.js index 9573d4c3..daca8d6a 100644 --- a/open-sse/utils/stream.js +++ b/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) { diff --git a/tests/unit/ollama-stream-tail.test.js b/tests/unit/ollama-stream-tail.test.js new file mode 100644 index 00000000..1f6bdb7e --- /dev/null +++ b/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'); + }); +});