fix(reasoning): sanitize streamed K3 think tags

open-sse/handlers/responseSanitizer/reasoning.ts

@@ -130,7 +130,7 @@ export function isTextualReasoningTagNativeRoute(providerId: string, modelId: st
     /deepseek[-_/]?r1\b/.test(routeId) ||
     /r1[-_/]?distill\b/.test(routeId) ||
     /(?:^|[/:_-])qwq(?:[/._:-]|$)/.test(routeId) ||
-    (providerId === "kimi-coding" && /(?:^|[/_-])k3(?:[/._:-]|$)/.test(modelId)) ||
+    /(?:^|[/_-])k3(?:[/._:-]|$)/.test(modelId) ||
     // 9router#2231: MiniMax M3 leaks raw <think>...</think> into `content` on its
     // OpenAI-format provider tiers (trae, huggingchat, bazaarlink, ollama-cloud,
     // opencode, cline, opencode-zen, codebuddy-cn). The direct minimax/minimax-cn

open-sse/utils/stream.ts

@@ -19,6 +19,7 @@ import {
   formatSSE,
   unwrapGeminiChunk,
   appendBoundedText,
+  buildSyntheticChatChunk,
   hasActiveDeltaValue,
 } from "./streamHelpers.ts";
 import { calculateCost } from "@/lib/usage/costCalculator";
@@ -61,6 +62,7 @@ import {
   getUnsupportedReasoningValue,
   hasUnsupportedReasoningSignal,
 } from "./reasoningFields.ts";
+import { applyThinkTag, flushThink, initThinkState } from "./thinkTagParser.ts";
 
 /**
  * Race a response body read against a timeout.
@@ -744,6 +746,7 @@ export function createSSEStream(options: StreamOptions = {}) {
   let passthroughToolCallSeq = 0;
   const allowedToolNames = extractAllowedToolNames(body);
   let skipPassthroughEvent = false;
+  const thinkState = initThinkState(mode === STREAM_MODE.PASSTHROUGH, provider, model);
 
   // State for translate mode (accumulatedContent for call log response body)
   const state: TranslateState | null =
@@ -1728,6 +1731,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                   let textualToolCallConverted = false;
                   let toolCallIdCoerced = false;
                   let splitMixedReasoningContent = false;
+                  const thinkParsed = applyThinkTag(thinkState, delta);
 
                   // Split combined reasoning+content deltas into separate SSE events.
                   // Standard OpenAI streaming never mixes both fields in one delta;
@@ -1767,6 +1771,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                   // to avoid blocking subsequent finish_reason / usage mutations)
                   const needsReserialization =
                     splitMixedReasoningContent ||
+                    thinkParsed ||
                     hadReasoningAlias ||
                     (delta?.content === "" && delta?.reasoning_content);
 
@@ -2360,21 +2365,9 @@ export function createSSEStream(options: StreamOptions = {}) {
                 };
                 flushOutput = `data: ${JSON.stringify(syntheticChunk)}\n\n`;
               } else {
-                const syntheticChunk = {
-                  id: passthroughResponsesId || `chatcmpl-${Date.now()}`,
-                  object: "chat.completion.chunk",
-                  created: Math.floor(Date.now() / 1000),
-                  model: model || "unknown",
-                  choices: [
-                    {
-                      index: 0,
-                      delta: {
-                        content: passthroughBufferedTextualToolCallContent,
-                      },
-                      finish_reason: null,
-                    },
-                  ],
-                };
+                const syntheticChunk = buildSyntheticChatChunk(passthroughResponsesId, model, {
+                  content: passthroughBufferedTextualToolCallContent,
+                });
                 flushOutput = `data: ${JSON.stringify(syntheticChunk)}\n\n`;
               }
               reqLogger?.appendConvertedChunk?.(flushOutput);
@@ -2386,6 +2379,18 @@ export function createSSEStream(options: StreamOptions = {}) {
               passthroughBufferedTextualToolCallContent = "";
             }
 
+            const accR = passthroughAccumulatedReasoning;
+            const accC = passthroughAccumulatedContent;
+            const thinkFlush = flushThink(thinkState, passthroughResponsesId, accR, accC);
+            if (thinkFlush) {
+              passthroughAccumulatedReasoning = thinkFlush.reasoning;
+              passthroughAccumulatedContent = thinkFlush.content;
+              totalContentLength += thinkFlush.addedLength;
+              clientPayloadCollector.push(thinkFlush.syntheticChunk);
+              reqLogger?.appendConvertedChunk?.(thinkFlush.flushOutput);
+              controller.enqueue(encoder.encode(thinkFlush.flushOutput));
+            }
+
             // Estimate usage if provider didn't return valid usage
             if (!hasValidUsage(usage) && totalContentLength > 0) {
               usage = estimateUsage(body, totalContentLength, sourceFormat || FORMATS.OPENAI);
@@ -2409,19 +2414,12 @@ export function createSSEStream(options: StreamOptions = {}) {
               // (pi CLI) reject the stream with "Stream ended without finish_reason".
               // Synthesize a terminal chunk when the upstream omitted one.
               if (shouldEmitDoneTerminator && !passthroughSawFinishReason) {
-                const syntheticFinishChunk = {
-                  id: passthroughResponsesId || `chatcmpl-${Date.now()}`,
-                  object: "chat.completion.chunk",
-                  created: Math.floor(Date.now() / 1000),
-                  model: model || "unknown",
-                  choices: [
-                    {
-                      index: 0,
-                      delta: {},
-                      finish_reason: passthroughHasToolCalls ? "tool_calls" : "stop",
-                    },
-                  ],
-                };
+                const syntheticFinishChunk = buildSyntheticChatChunk(
+                  passthroughResponsesId,
+                  model,
+                  {},
+                  passthroughHasToolCalls ? "tool_calls" : "stop"
+                );
                 const finishOutput = `data: ${JSON.stringify(syntheticFinishChunk)}\n\n`;
                 reqLogger?.appendConvertedChunk?.(finishOutput);
                 controller.enqueue(encoder.encode(finishOutput));

open-sse/utils/streamHelpers.ts

@@ -475,6 +475,28 @@ export function formatSSE(data: unknown, sourceFormat: string): string {
   return `data: ${JSON.stringify(data)}\n\n`;
 }
 
+/**
+ * Build a synthetic OpenAI-shaped chat completion chunk for a manually
+ * assembled SSE delta (end-of-stream flushes, textual tool-call fallback,
+ * think-tag reasoning flush, terminal finish_reason synthesis, etc). Reuses
+ * the same `id`/`created` fallback and `choices[0]` shape every passthrough
+ * flush site needs.
+ */
+export function buildSyntheticChatChunk(
+  responsesId: string | null | undefined,
+  model: string | null | undefined,
+  delta: Record<string, unknown>,
+  finishReason: string | null = null
+): Record<string, unknown> {
+  return {
+    id: responsesId || `chatcmpl-${Date.now()}`,
+    object: "chat.completion.chunk",
+    created: Math.floor(Date.now() / 1000),
+    model: model || "unknown",
+    choices: [{ index: 0, delta, finish_reason: finishReason }],
+  };
+}
+
 const STREAM_SUMMARY_TEXT_LIMIT = 64 * 1024;
 
 // Bounded accumulator for streamed content/reasoning text — caps memory on long streams

open-sse/utils/thinkTagParser.ts

@@ -15,9 +15,49 @@
  *   // content = "final answer..."
  */
 
+import { shouldParseTextualReasoningTags } from "../handlers/responseSanitizer/reasoning.ts";
+import { appendBoundedText, buildSyntheticChatChunk } from "./streamHelpers.ts";
+
 const THINK_OPEN = "<think>";
 const THINK_CLOSE = "</think>";
 
+/**
+ * Create the mutable streaming-parse context for one SSE stream.
+ * `enabled` decides whether the caller should attempt think-tag parsing at
+ * all for this stream (passthrough mode + a provider/model that is known to
+ * emit textual `<think>` tags); `active`/`insideThink`/`buffer` track parse
+ * progress once a tag has been seen. `model` is stashed for `flushThink`'s
+ * synthetic chunk so callers don't need to pass it again at flush time.
+ *
+ * @param {boolean} isPassthroughMode
+ * @param {unknown} provider
+ * @param {string} [model]
+ */
+export function initThinkState(isPassthroughMode: boolean, provider?: unknown, model?: string) {
+  return {
+    enabled: isPassthroughMode && shouldParseTextualReasoningTags(provider, model),
+    insideThink: false,
+    buffer: "",
+    active: false,
+    model,
+  };
+}
+
+/**
+ * Check whether a streaming text chunk contains a full `<think>` open tag,
+ * or ends with a prefix of one (so the caller knows to keep buffering
+ * instead of treating the tag as ordinary content split across chunks).
+ *
+ * @param {string} value - Streaming text chunk
+ * @returns {boolean}
+ */
+export function containsOrMayEndWithThinkOpenTag(value: string): boolean {
+  return (
+    value.includes(THINK_OPEN) ||
+    ["<", "<t", "<th", "<thi", "<thin"].some((suffix) => value.endsWith(suffix))
+  );
+}
+
 /**
  * Check if text contains think tags
  * @param {string} text - Raw output text
@@ -126,6 +166,33 @@ export function processStreamingThinkDelta(delta, ctx) {
   };
 }
 
+/**
+ * Parse a streaming SSE delta's `content` field for `<think>` tags in place,
+ * when applicable. No-ops (returns false, delta untouched) unless parsing is
+ * `ctx.enabled` for this stream AND either a tag is already open or this
+ * chunk contains/may end with a `<think>` open tag.
+ *
+ * Mutates `delta.content` (stripped of think markup) and sets
+ * `delta.reasoning_content` when reasoning text was extracted.
+ *
+ * @param {object} ctx - Mutable context object from `initThinkState`
+ * @param {{ content: unknown, reasoning_content?: string }} delta
+ * @returns {boolean} true if the delta was parsed for think tags
+ */
+export function applyThinkTag(
+  ctx,
+  delta: { content: unknown; reasoning_content?: string }
+): boolean {
+  if (!ctx.enabled || typeof delta?.content !== "string") return false;
+  if (!ctx.active && !containsOrMayEndWithThinkOpenTag(delta.content)) return false;
+
+  ctx.active = true;
+  const { reasoningDelta, contentDelta } = processStreamingThinkDelta(delta.content, ctx);
+  delta.content = contentDelta || "";
+  if (reasoningDelta) delta.reasoning_content = reasoningDelta;
+  return true;
+}
+
 /**
  * Flush remaining buffer content from streaming context.
  * Call this when the stream ends.
@@ -144,3 +211,58 @@ export function flushThinkBuffer(ctx) {
   }
   return { reasoningDelta: null, contentDelta: remaining || null };
 }
+
+/**
+ * Flush remaining buffer content from a streaming context at end-of-stream
+ * and assemble a synthetic OpenAI-shaped chat completion chunk, plus the
+ * accumulator updates for it, ready for the caller to enqueue as SSE output.
+ *
+ * Returns `null` when parsing was never `ctx.enabled`/`ctx.active` for this
+ * stream, or when there is nothing left to flush — callers can skip
+ * emitting an empty synthetic chunk and leave their accumulators untouched.
+ *
+ * @param {object} ctx - Mutable context object from `initThinkState`
+ *   (its `model` is reused for the synthetic chunk)
+ * @param {string} [responsesId] - Response id to reuse, falls back to a
+ *   generated `chatcmpl-<timestamp>` id
+ * @param {string} accReasoning - Caller's running reasoning accumulator
+ * @param {string} accContent - Caller's running content accumulator
+ * @returns {{
+ *   syntheticChunk: Record<string, unknown>,
+ *   flushOutput: string,
+ *   reasoning: string,
+ *   content: string,
+ *   addedLength: number
+ * }|null}
+ */
+export function flushThink(
+  ctx,
+  responsesId: string | undefined,
+  accReasoning: string,
+  accContent: string
+): {
+  syntheticChunk: Record<string, unknown>;
+  flushOutput: string;
+  reasoning: string;
+  content: string;
+  addedLength: number;
+} | null {
+  if (!ctx.enabled || !ctx.active) return null;
+
+  const { reasoningDelta, contentDelta } = flushThinkBuffer(ctx);
+  if (!reasoningDelta && !contentDelta) return null;
+
+  const delta: Record<string, string> = {};
+  if (reasoningDelta) delta.reasoning_content = reasoningDelta;
+  if (contentDelta) delta.content = contentDelta;
+
+  const syntheticChunk = buildSyntheticChatChunk(responsesId, ctx.model, delta);
+
+  return {
+    syntheticChunk,
+    flushOutput: `data: ${JSON.stringify(syntheticChunk)}\n\n`,
+    reasoning: appendBoundedText(accReasoning, reasoningDelta || ""),
+    content: appendBoundedText(accContent, contentDelta || ""),
+    addedLength: (reasoningDelta?.length || 0) + (contentDelta?.length || 0),
+  };
+}

tests/unit/responsesanitizer-reasoning-split.test.ts

@@ -65,6 +65,19 @@ describe("responseSanitizer/reasoning — Kimi Code K3 textual reasoning-tag rou
     assert.equal(delta.content, "final answer");
     assert.equal(delta.reasoning_content, "reasoning here");
   });
+
+  it("extracts <think> blocks from a K3 route", () => {
+    const chunk = {
+      choices: [{ index: 0, delta: { content: "<think>reasoning here</think>final answer" } }],
+    };
+    const sanitized = sanitizeOpenAIResponse(chunk, {
+      parseTextualReasoningTags: shouldParseTextualReasoningTags("kimi-coding", "k3"),
+    }) as { choices: Array<{ delta: { content: string; reasoning_content?: string } }> };
+
+    const delta = sanitized.choices[0].delta;
+    assert.equal(delta.content, "final answer");
+    assert.equal(delta.reasoning_content, "reasoning here");
+  });
 });
 
 // ── MiniMax M3 textual reasoning-tag route (9router#2231) ──────────────────────

tests/unit/stream-utilities.test.ts

@@ -91,6 +91,57 @@ test("createPassthroughStreamWithLogger omits [DONE] for Responses clients", asy
   assert.doesNotMatch(result, /data: \[DONE\]/);
 });
 
+test("createPassthroughStreamWithLogger separates K3 thinking tags", async () => {
+  const transform = createPassthroughStreamWithLogger(
+    "kimi-coding",
+    null,
+    null,
+    "k3",
+    null,
+    null,
+    null,
+    null,
+    null,
+    "openai"
+  );
+  const writer = transform.writable.getWriter();
+  const encoder = new TextEncoder();
+
+  await writer.write(
+    encoder.encode(
+      'data: {"id":"chatcmpl_1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"<think>reasoning"},"finish_reason":null}]}\n\n'
+    )
+  );
+  await writer.write(
+    encoder.encode(
+      'data: {"id":"chatcmpl_1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"content":"</think>final answer"},"finish_reason":null}]}\n\n'
+    )
+  );
+  await writer.close();
+
+  const reader = transform.readable.getReader();
+  const decoder = new TextDecoder();
+  let result = "";
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    result += decoder.decode(value);
+  }
+
+  const payloads = result
+    .split("\n")
+    .filter((line) => line.startsWith("data: {") && line !== "data: [DONE]")
+    .map((line) => JSON.parse(line.slice(6)));
+  const visible = payloads.map((payload) => payload.choices?.[0]?.delta?.content || "").join("");
+  const reasoning = payloads
+    .map((payload) => payload.choices?.[0]?.delta?.reasoning_content || "")
+    .join("");
+
+  assert.equal(visible, "final answer");
+  assert.equal(reasoning, "reasoning");
+  assert.doesNotMatch(result, /<\/?think>/);
+});
+
 test("createPassthroughStreamWithLogger synthesizes reasoning summary events from reasoning output items", async () => {
   const transform = createPassthroughStreamWithLogger(
     "codex",