fix(sse): resume stream recovery after a clean stop with reasoning-only output (#11151)

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

@@ -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<Uint8Array>) => {
     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;

tests/unit/stream-continuation-wiring.test.ts

@@ -46,6 +46,10 @@ async function collectText(stream: ReadableStream<Uint8Array>): Promise<string>
 
 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.");
+});

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", () => {