diff --git a/changelog.d/fixes/13620-combo-reasoning-only-sse-burst.md b/changelog.d/fixes/13620-combo-reasoning-only-sse-burst.md new file mode 100644 index 0000000000..fa54e1cbec --- /dev/null +++ b/changelog.d/fixes/13620-combo-reasoning-only-sse-burst.md @@ -0,0 +1 @@ +- **fix(sse):** stream reasoning deltas from combo targets incrementally instead of buffering them into a single burst, and stop rejecting reasoning-only streams as an empty completion (#13620) — thanks @NaNomicon diff --git a/open-sse/utils/streamHelpers.ts b/open-sse/utils/streamHelpers.ts index 39aafbacf9..13a57db6af 100644 --- a/open-sse/utils/streamHelpers.ts +++ b/open-sse/utils/streamHelpers.ts @@ -202,11 +202,7 @@ export function createSSEDataLineNormalizer(): SSEDataLineNormalizer { const normalizedLine = line.replace(CR_STRIP_RE, ""); const trimmed = normalizedLine.trim(); - if ( - trimmed && - SSE_FIELD_RE.test(trimmed) && - hasSelfDescribingPendingDataPayload() - ) { + if (trimmed && SSE_FIELD_RE.test(trimmed) && hasSelfDescribingPendingDataPayload()) { flush(output); } @@ -220,7 +216,9 @@ export function createSSEDataLineNormalizer(): SSEDataLineNormalizer { }; } -export function createSSEEventPrefixBuffer(options?: { forwardEvent?: boolean }): SSEEventPrefixBuffer { +export function createSSEEventPrefixBuffer(options?: { + forwardEvent?: boolean; +}): SSEEventPrefixBuffer { let lines: string[] = []; let emitted = false; // The `event:` line is only part of the SSE framing for protocols that define @@ -277,12 +275,7 @@ function hasOpenAICompatibleStreamValue(parsed: Record): boolea const delta = isRecord(choice.delta) ? choice.delta : null; if (!delta) return false; if (typeof delta.content === "string" && delta.content.length > 0) return true; - if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) { - return true; - } - if (typeof delta.reasoning_text === "string" && delta.reasoning_text.length > 0) { - return true; - } + if (hasAnyReasoningSignal(delta)) return true; return Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0; }); } diff --git a/tests/unit/issue-13620-combo-reasoning-only-sse-burst.test.ts b/tests/unit/issue-13620-combo-reasoning-only-sse-burst.test.ts new file mode 100644 index 0000000000..59613d2ce0 --- /dev/null +++ b/tests/unit/issue-13620-combo-reasoning-only-sse-burst.test.ts @@ -0,0 +1,126 @@ +/** + * Issue #13620 — combo quality-gate bounded peek buffers bare `delta.reasoning` + * SSE chunks (Ollama Cloud shape) because `hasOpenAICompatibleStreamValue()` + * (open-sse/utils/streamHelpers.ts) only recognizes `content`, + * `reasoning_content`, `reasoning_text`, and `tool_calls` — not bare + * `reasoning`. As a result the peek loop never exits on a reasoning-only + * chunk and keeps buffering every reasoning delta until the first real + * `content` delta arrives (client sees dead silence, then a single burst), + * and a stream that finishes with reasoning-only content is rejected as an + * "empty completion" (secondary 502 failure mode). + */ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { isKnownNonClaudeStreamPayload } = await import("../../open-sse/utils/streamHelpers.ts"); +const { validateResponseQuality } = await import("../../open-sse/services/combo.ts"); + +const encoder = new TextEncoder(); +const silentLog = { warn: () => {} }; + +test("#13620 root cause: isKnownNonClaudeStreamPayload must recognize bare delta.reasoning (Ollama Cloud shape)", () => { + const parsed = { + choices: [ + { index: 0, delta: { role: "assistant", content: "", reasoning: "We" }, finish_reason: null }, + ], + }; + const recognized = isKnownNonClaudeStreamPayload(parsed); + assert.equal( + recognized, + true, + "bare delta.reasoning must be recognized as stream content so the combo peek stops buffering immediately, " + + "matching hasValuableContent()/hasAnyReasoningSignal() which already treat it as real signal" + ); +}); + +// Emits ONE physical Uint8Array chunk per SSE frame (mirrors real network +// framing, unlike a single enqueue() of the whole body) and counts how many +// chunks the underlying stream has handed out by the time the caller stops +// reading — this is what actually distinguishes "peek recognizes the first +// reasoning delta and stops buffering there" from "peek keeps consuming +// (buffering) every reasoning chunk until the content chunk arrives". +function makeReasoningOnlyThenContentStream(): { + response: Response; + chunksPulled: { count: number }; +} { + const frames = [ + { choices: [{ index: 0, delta: { role: "assistant", content: "", reasoning: "We" } }] }, + { choices: [{ index: 0, delta: { content: "", reasoning: " need" } }] }, + { choices: [{ index: 0, delta: { content: "", reasoning: " to think." } }] }, + { choices: [{ index: 0, delta: { content: "Hello" } }] }, + { choices: [{ index: 0, delta: {}, finish_reason: "stop" }] }, + ].map((c) => encoder.encode(`data: ${JSON.stringify(c)}\n\n`)); + frames.push(encoder.encode("data: [DONE]\n\n")); + + const chunksPulled = { count: 0 }; + let idx = 0; + const stream = new ReadableStream({ + pull(controller) { + if (idx >= frames.length) { + controller.close(); + return; + } + controller.enqueue(frames[idx++]); + chunksPulled.count = idx; + }, + }); + return { + response: new Response(stream, { + status: 200, + headers: { "content-type": "text/event-stream" }, + }), + chunksPulled, + }; +} + +test("#13620 behavioral: peek must stop buffering at the first reasoning delta, not wait for content", async () => { + const { response, chunksPulled } = makeReasoningOnlyThenContentStream(); + const out = await validateResponseQuality(response, true, silentLog); + assert.equal(out.valid, true, `expected valid, got reason: ${out.reason}`); + assert.ok(out.clonedResponse, "expected a clonedResponse to replay/forward the stream"); + + // Frame indices: 0-2 = reasoning-only deltas, 3 = content, 4 = finish_reason, 5 = [DONE]. + // Correct: the peek recognizes frame 0 immediately via a single reader.read() call. + // The WHATWG ReadableStream spec itself speculatively pulls one extra chunk into its + // internal queue right after that read() resolves (default highWaterMark backpressure + // refill) — so up to 2 physical chunks may be pulled even when the peek loop performs + // exactly one logical read. That prefetch is unavoidable engine behavior, not something + // validateResponseQuality controls, so the threshold below allows for it. + // Bug: the peek loop itself keeps calling reader.read() through frames 0-2 (each + // triggering its own follow-up prefetch), only stopping at frame 3 -> 4-5 chunks pulled. + assert.ok( + chunksPulled.count <= 2, + "expected the peek to stop after at most 1 logical read (<=2 chunks pulled, accounting for " + + `ReadableStream's own one-chunk-ahead prefetch); actual: ${chunksPulled.count} chunks were ` + + "pulled before the peek recognized any content — this is the reasoning-prefix burst-buffering bug" + ); +}); + +function makeReasoningOnlyTerminatedStream(): Response { + const chunks = [ + { choices: [{ index: 0, delta: { role: "assistant", reasoning: "Deep" } }] }, + { choices: [{ index: 0, delta: { reasoning: " thinking..." } }] }, + { choices: [{ index: 0, delta: {}, finish_reason: "stop" }] }, + ]; + const body = chunks.map((c) => `data: ${JSON.stringify(c)}\n\n`).join("") + "data: [DONE]\n\n"; + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(body)); + controller.close(); + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } } + ); +} + +test("#13620 secondary failure mode: reasoning-only stream that terminates cleanly must not be rejected as an empty completion", async () => { + const res = makeReasoningOnlyTerminatedStream(); + const out = await validateResponseQuality(res, true, silentLog); + assert.equal( + out.valid, + true, + `expected valid (reasoning-only is real output, mirrors the non-streaming validator's ` + + `reasoning_content ?? reasoning acceptance) — got reason: ${out.reason}` + ); +});