From 80b8d2a8a284baff3e5740d0a086f1c68ff39531 Mon Sep 17 00:00:00 2001 From: Dizzle <112548150+maxmad64bis@users.noreply.github.com> Date: Sun, 23 Aug 2026 02:54:48 +0200 Subject: [PATCH] fix(sse): resume stream recovery after a clean stop with reasoning-only output (#11151) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merged after conflict resolution against the tip's #11109 (per-call tool_call tracking): scanOpenAiSseText keeps the per-call finish_reason special-case AND gains reasoningText + literal finishReason; canContinue uses the in-flight predicate with the new reasoning-only-clean-stop escape. One integration fix on the branch: the PR's hallucinatedEmptyStop referenced emittedToolCall, which #11109 had renamed — the branch now tracks emittedSawToolCall at the emitted level (any tool_call delta, complete or not), preserving the PR's don't-recover-after-tool-calls intent. Chain suites green: stream-continuation-wiring + stream-continuation + stream-recovery-toolcall 29/29. Thank you @maxmad64bis! --- open-sse/services/streamRecovery.ts | 105 +++++++++++++++--- tests/unit/stream-continuation-wiring.test.ts | 92 +++++++++++++++ tests/unit/stream-continuation.test.ts | 33 ++++++ 3 files changed, 214 insertions(+), 16 deletions(-) diff --git a/open-sse/services/streamRecovery.ts b/open-sse/services/streamRecovery.ts index 37a4867f13..129039c26c 100644 --- a/open-sse/services/streamRecovery.ts +++ b/open-sse/services/streamRecovery.ts @@ -183,6 +183,11 @@ export function hasTerminalMarker(bytes: Uint8Array): boolean { export interface OpenAiSseScan { /** Concatenated assistant text seen across `choices[].delta.content`. */ text: string; + /** Concatenated reasoning trace seen across `choices[].delta.reasoning_content`. Some + * providers stream the entire answer here and leave `content` empty/null — tracked + * separately so a clean stop with reasoning-only output can still be recognized as + * "nothing usable was delivered" instead of "a normal empty turn". */ + reasoningText: string; /** True if any `choices[].delta.tool_calls` appeared — NEVER continue those. */ sawToolCall: boolean; /** @@ -201,6 +206,10 @@ export interface OpenAiSseScan { * (and the client-visible SSE) is still eligible to be resumed past it. */ terminal: boolean; + /** The literal `finish_reason` string when present (e.g. "stop", "tool_calls", "length", + * "content_filter"), or `null` if none was seen. `terminal` alone is not precise enough + * to gate the reasoning-only-stop continuation — it must fire on `"stop"` only. */ + finishReason: string | null; /** True if at least one OpenAI-shaped `choices[].delta` was parsed (format gate). */ parsedOpenAi: boolean; } @@ -212,12 +221,22 @@ export interface OpenAiSseScan { */ export function scanOpenAiSseText(sse: string): OpenAiSseScan { let text = ""; + let reasoningText = ""; let sawToolCall = false; let toolCallFinished = false; let terminal = false; + let finishReason: string | null = null; let parsedOpenAi = false; if (typeof sse !== "string" || sse.length === 0) { - return { text, sawToolCall, sawToolCallInFlight: false, terminal, parsedOpenAi }; + return { + text, + reasoningText, + sawToolCall, + sawToolCallInFlight: false, + terminal, + finishReason, + parsedOpenAi, + }; } for (const line of sse.split("\n")) { const trimmed = line.trimStart(); @@ -242,21 +261,33 @@ export function scanOpenAiSseText(sse: string): OpenAiSseScan { parsedOpenAi = true; const content = (delta as { content?: unknown }).content; if (typeof content === "string") text += content; + const reasoning = (delta as { reasoning_content?: unknown }).reasoning_content; + if (typeof reasoning === "string") reasoningText += reasoning; const toolCalls = (delta as { tool_calls?: unknown }).tool_calls; if (Array.isArray(toolCalls) && toolCalls.length > 0) sawToolCall = true; } - const finishReason = (choice as { finish_reason?: unknown })?.finish_reason; - if (finishReason === "tool_calls") { + const rawFinishReason = (choice as { finish_reason?: unknown })?.finish_reason; + if (rawFinishReason === "tool_calls") { // Ends this one choice, but the overall stream/turn stays continuable — // never counts as the general terminal marker (see OpenAiSseScan.terminal). toolCallFinished = true; - } else if (finishReason != null) { + finishReason = "tool_calls"; + } else if (rawFinishReason != null) { terminal = true; + if (typeof rawFinishReason === "string") finishReason = rawFinishReason; } } } const sawToolCallInFlight = sawToolCall && !toolCallFinished; - return { text, sawToolCall, sawToolCallInFlight, terminal, parsedOpenAi }; + return { + text, + reasoningText, + sawToolCall, + sawToolCallInFlight, + terminal, + finishReason, + parsedOpenAi, + }; } export interface ContinuableBody { @@ -267,8 +298,10 @@ export interface ContinuableBody { /** * Build a re-request body that continues from `assistantSoFar` by appending it as an - * assistant turn. Returns null when the body has no `messages` array or the partial text - * is empty (nothing to continue from). Does not mutate the original. + * assistant turn. When `assistantSoFar` is empty (nothing usable was emitted yet — e.g. a + * clean stop that only produced reasoning), the messages are re-sent unchanged instead of + * appending an empty assistant turn: this simply re-asks for a real answer. Returns null + * only when the body has no `messages` array at all (nothing to continue from). */ export function makeContinuationBody( body: ContinuableBody, @@ -276,10 +309,13 @@ export function makeContinuationBody( ): (ContinuableBody & { messages: unknown[] }) | null { if (!body || typeof body !== "object") return null; if (!Array.isArray(body.messages) || body.messages.length === 0) return null; - if (typeof assistantSoFar !== "string" || assistantSoFar.length === 0) return null; + if (typeof assistantSoFar !== "string") return null; return { ...body, - messages: [...body.messages, { role: "assistant", content: assistantSoFar }], + messages: + assistantSoFar.length > 0 + ? [...body.messages, { role: "assistant", content: assistantSoFar }] + : [...body.messages], stream: true, }; } @@ -390,8 +426,13 @@ export function createRecoverableStream( let continuations = 0; let emittedTail = ""; // raw SSE not yet scanned (awaiting an event boundary) let emittedText = ""; // assistant text already delivered to the client + let emittedReasoningText = ""; // reasoning trace already delivered (never shown to the client, + // tracked only to distinguish "a real empty turn" from "the whole + // answer stayed in the reasoning channel") + let emittedFinishReason: string | null = null; // literal finish_reason last seen, if any let emittedTerminal = false; let emittedToolCallInFlight = false; + let emittedSawToolCall = false; // any tool_call delta seen, complete or not let emittedParsedOpenAi = false; // Enqueue to the client and, when continuation is enabled, fold the chunk into the @@ -409,8 +450,11 @@ export function createRecoverableStream( emittedTail = emittedTail.slice(boundary + 2); const scan = scanOpenAiSseText(complete); emittedText += scan.text; + emittedReasoningText += scan.reasoningText; + if (scan.finishReason !== null) emittedFinishReason = scan.finishReason; if (scan.terminal) emittedTerminal = true; if (scan.sawToolCallInFlight) emittedToolCallInFlight = true; + if (scan.sawToolCall) emittedSawToolCall = true; if (scan.parsedOpenAi) emittedParsedOpenAi = true; }; @@ -418,15 +462,42 @@ export function createRecoverableStream( for (const chunk of holdback.flush()) emit(controller, chunk); }; - // A post-commit truncation is continuable only for a plain-text OpenAI-compatible - // stream that has not finished and has no tool call in flight. + // A post-commit truncation is continuable for a plain-text OpenAI-compatible stream that + // has no tool call in flight, AND either: + // - has not finished yet (the original #4131 truncation case), or + // - finished with a literal finish_reason of "stop" but delivered nothing usable while a + // non-empty reasoning trace shows the provider spent its whole turn "thinking" and never + // turned that into an answer (some providers put the entire response in + // reasoning_content and leave content empty). Gated on the LITERAL "stop" value, not the + // generic `terminal` flag — `terminal` also covers "length"/"content_filter"/a bare + // [DONE], which are out of scope for this specific recovery. + // + // Known consequence of the hallucinatedEmptyStop path (flagged in cross-review, accepted as + // inherent to tryContinue's existing design, not new to this fix): the original upstream's + // `finish_reason:"stop"` chunk was already forwarded to the client via `emit()`'s unconditional + // `controller.enqueue(chunk)` (streamRecovery.ts:381) BEFORE this scan ever runs — that is how + // `emittedFinishReason`/`emittedTerminal` get set in the first place. So the client sees an + // empty "stop" marker from the original turn, then — once the continuation succeeds — the real + // answer plus a SECOND `emitCleanTerminal` from `tryContinue`. This mirrors what already + // happens for the pre-existing truncation-continuation case (a truncated stream can likewise + // have partially delivered SSE framing before `tryContinue` appends more); it is not a new + // double-close of the underlying `ReadableStream` (`controller.close()` runs exactly once, + // after `tryContinue` returns). An SSE client that treats a bare `finish_reason:"stop"` as an + // unconditional end-of-turn (rather than waiting for `[DONE]`) may need updating separately — + // out of scope for this fix, which targets the observed opencode/OmniRoute pairing where the + // client kept the connection open. + const hallucinatedEmptyStop = () => + emittedFinishReason === "stop" && + !emittedSawToolCall && + emittedText.length === 0 && + emittedReasoningText.length > 0; + const canContinue = () => continueEnabled && continuations < maxContinuations && emittedParsedOpenAi && !emittedToolCallInFlight && - !emittedTerminal && - emittedText.length > 0; + (emittedText.length > 0 ? !emittedTerminal : hallucinatedEmptyStop()); const emitCleanTerminal = (controller: ReadableStreamDefaultController) => { controller.enqueue( @@ -527,9 +598,11 @@ export function createRecoverableStream( const { done, value } = result; if (done) { if (holdback.committed) { - // Graceful end after commit: if it lacks a terminal marker it is a silent - // truncation — try to continue; otherwise (clean finish) just close. - if (!emittedTerminal && (await tryContinue(controller))) { + // Graceful end after commit: try a mid-stream continuation whenever canContinue() + // says the stream is worth continuing (silent truncation, or a clean-but-empty + // reasoning-only stop) — canContinue() is the single source of truth here, same as + // the read-error branch above. + if (await tryContinue(controller)) { runFinalize(); controller.close(); return; diff --git a/tests/unit/stream-continuation-wiring.test.ts b/tests/unit/stream-continuation-wiring.test.ts index 2f246e325d..1081d6ea9b 100644 --- a/tests/unit/stream-continuation-wiring.test.ts +++ b/tests/unit/stream-continuation-wiring.test.ts @@ -46,6 +46,10 @@ async function collectText(stream: ReadableStream): Promise const ROLE = 'data: {"choices":[{"delta":{"role":"assistant"}}]}\n\n'; const content = (s: string) => `data: {"choices":[{"delta":{"content":${JSON.stringify(s)}}}]}\n\n`; +const reasoning = (s: string) => + `data: {"choices":[{"delta":{"reasoning_content":${JSON.stringify(s)}}}]}\n\n`; +const finishStopNoContent = 'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n'; +const finishLengthNoContent = 'data: {"choices":[{"delta":{},"finish_reason":"length"}]}\n\n'; test("mid-stream continuation: stitches the suffix after a silent post-commit truncation", async () => { // Commits on chunk 1, emits "Hello wor", then ends WITHOUT a terminal marker (silent cut). @@ -115,3 +119,91 @@ test("tool-call in flight is never continued (would corrupt tool JSON)", async ( await collectText(stream); assert.equal(continued, false, "continuation must NOT fire once a tool call has started streaming"); }); + +test("mid-stream continuation: a clean stop with reasoning-only output (no answer) triggers a continuation", async () => { + const initial = streamFrom([ + ROLE, + reasoning("the model thinks through the problem here..."), + finishStopNoContent, + ]); + let continueArg = "__unset__"; + const stream = createRecoverableStream(initial, async () => null, { + finalize: () => {}, + now: steppingClock(), + continueStream: async (soFar: string) => { + continueArg = soFar; + return streamFrom([content("Here is the actual answer."), "data: [DONE]\n\n"]); + }, + }); + const out = await collectText(stream); + const scan = scanOpenAiSseText(out); + assert.equal(continueArg, "", "nothing usable was emitted — the re-request has an empty prefill"); + assert.equal( + scan.text, + "Here is the actual answer.", + "the client gets a real answer instead of silence" + ); + assert.equal(scan.terminal, true); +}); + +test("mid-stream continuation: a clean stop with truly empty output (no text, no reasoning) is left unchanged", async () => { + const initial = streamFrom([ROLE, finishStopNoContent]); + let continued = false; + const stream = createRecoverableStream(initial, async () => null, { + finalize: () => {}, + now: steppingClock(), + continueStream: async () => { + continued = true; + return streamFrom([content("nope"), "data: [DONE]\n\n"]); + }, + }); + await collectText(stream); + assert.equal( + continued, + false, + "no reasoning trace means there is nothing to act on — do not guess" + ); +}); + +test("mid-stream continuation: finish_reason 'length' with reasoning-only output does NOT trigger a continuation", async () => { + // Regression guard for a blocker found in cross-review: widening the gate to any + // terminal marker (instead of the literal finish_reason "stop") would wrongly spend a + // continuation attempt on a token-limit cutoff, which is out of this fix's scope. + const initial = streamFrom([ + ROLE, + reasoning("the model was still thinking when it hit the token limit..."), + finishLengthNoContent, + ]); + let continued = false; + const stream = createRecoverableStream(initial, async () => null, { + finalize: () => {}, + now: steppingClock(), + continueStream: async () => { + continued = true; + return streamFrom([content("nope"), "data: [DONE]\n\n"]); + }, + }); + await collectText(stream); + assert.equal(continued, false, "finish_reason 'length' is out of scope for this fix"); +}); + +test("mid-stream continuation: real content alongside reasoning at a clean stop is left unchanged (non-regression)", async () => { + const initial = streamFrom([ + ROLE, + reasoning("thinking..."), + content("The real answer."), + finishStopNoContent, + ]); + let continued = false; + const stream = createRecoverableStream(initial, async () => null, { + finalize: () => {}, + now: steppingClock(), + continueStream: async () => { + continued = true; + return streamFrom([content("nope"), "data: [DONE]\n\n"]); + }, + }); + const scan = scanOpenAiSseText(await collectText(stream)); + assert.equal(continued, false, "real content was delivered — nothing to recover"); + assert.equal(scan.text, "The real answer."); +}); diff --git a/tests/unit/stream-continuation.test.ts b/tests/unit/stream-continuation.test.ts index caffe936df..8a1da37ce9 100644 --- a/tests/unit/stream-continuation.test.ts +++ b/tests/unit/stream-continuation.test.ts @@ -21,6 +21,30 @@ test("scanOpenAiSseText accumulates content deltas and flags an OpenAI-compat st assert.equal(r.terminal, false); }); +test("scanOpenAiSseText accumulates reasoning_content deltas separately from content", () => { + const sse = + 'data: {"choices":[{"delta":{"role":"assistant"}}]}\n\n' + + 'data: {"choices":[{"delta":{"reasoning_content":"thinking..."}}]}\n\n' + + 'data: {"choices":[{"delta":{"reasoning_content":" more"}}]}\n\n'; + const r = scanOpenAiSseText(sse); + assert.equal(r.reasoningText, "thinking... more"); + assert.equal(r.text, "", "reasoning_content must never leak into the visible text field"); + assert.equal(r.parsedOpenAi, true); +}); + +test("scanOpenAiSseText captures the literal finish_reason value", () => { + const stop = scanOpenAiSseText('data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n'); + assert.equal(stop.finishReason, "stop"); + + const length = scanOpenAiSseText( + 'data: {"choices":[{"delta":{"content":"x"},"finish_reason":"length"}]}\n\n' + ); + assert.equal(length.finishReason, "length"); + + const none = scanOpenAiSseText('data: {"choices":[{"delta":{"content":"x"}}]}\n\n'); + assert.equal(none.finishReason, null, "no finish_reason seen means null, not a guessed default"); +}); + test("scanOpenAiSseText detects the terminal [DONE] marker", () => { const r = scanOpenAiSseText('data: {"choices":[{"delta":{"content":"hi"}}]}\n\ndata: [DONE]\n\n'); assert.equal(r.text, "hi"); @@ -65,6 +89,15 @@ test("makeContinuationBody refuses bodies without a messages array or empty text assert.equal(makeContinuationBody(null as never, "t"), null); }); +test("makeContinuationBody accepts an empty prefill by re-sending the messages unchanged", () => { + const body = { model: "x", stream: true, messages: [{ role: "user", content: "hi" }] }; + const out = makeContinuationBody(body, ""); + assert.ok(out, "an empty prefill must still produce a re-request body, not null"); + assert.equal(out!.messages.length, 1, "no empty assistant turn is appended"); + assert.deepEqual(out!.messages[0], { role: "user", content: "hi" }); + assert.equal(out!.stream, true); +}); + // ── trimContinuationOverlap ─────────────────────────────────────────────────── test("trimContinuationOverlap removes a duplicated seam so the join is append-only", () => {