feat(sse): Kiro inline <thinking> stream splitter (#4911)

Integrated into release/v3.8.38 (leva 5)

CHANGELOG.md

@@ -11,6 +11,7 @@ _In development — bullets added per PR; finalized at release._
 ### ✨ New Features
 
 - **feat(blackbox):** refresh provider model catalog with latest models. (thanks @ptkelanatechsolutions)
+- **kiro**: inline `<thinking>` stream splitter — when `<thinking_mode>enabled</thinking_mode>` is present, `assistantResponseEvent` content is now split into separate `delta.content` / `delta.reasoning_content` SSE chunks (new `open-sse/executors/kiroThinking.ts` module wired into `KiroExecutor.transformEventStreamToSSE`).
 
 ### 🔧 Bug Fixes
 

open-sse/executors/kiro.ts

@@ -8,6 +8,7 @@ import {
 import { PROVIDERS } from "../config/constants.ts";
 import { v4 as uuidv4 } from "uuid";
 import { refreshKiroToken } from "../services/tokenRefresh.ts";
+import { splitInlineThinking, flushPendingThinking, type KiroThinkingState } from "./kiroThinking.ts";
 
 type JsonRecord = Record<string, unknown>;
 
@@ -34,6 +35,10 @@ type KiroStreamState = {
   hasContextUsage?: boolean;
   hasMeteringEvent?: boolean;
   usage?: UsageSummary;
+  hasReasoningContent?: boolean;
+  reasoningChunkCount?: number;
+  // Inline-thinking splitter state (populated only when thinkingExpected=true).
+  thinking?: KiroThinkingState;
 };
 
 type EventFrame = {
@@ -338,18 +343,51 @@ export class KiroExecutor extends BaseExecutor {
       return { response, url, headers, transformedBody };
     }
 
-    // For Kiro, we need to transform the binary EventStream to SSE
-    // Create a TransformStream to convert binary to SSE text
-    const transformedResponse = this.transformEventStreamToSSE(response, model);
+    // For Kiro, we need to transform the binary EventStream to SSE.
+    // Create a TransformStream to convert binary to SSE text.
+    //
+    // When the user enabled thinking, Claude on Kiro streams its reasoning
+    // **inline** as `<thinking>…</thinking>` blocks inside
+    // `assistantResponseEvent.content` rather than as separate
+    // `reasoningContentEvent` frames. We pass a hint so the transform stream
+    // can split that inline reasoning into the OpenAI `delta.reasoning_content`
+    // channel.
+    const tb = transformedBody as Record<string, unknown>;
+    const userContent =
+      (
+        (
+          (
+            (tb?.conversationState as Record<string, unknown>)
+              ?.currentMessage as Record<string, unknown>
+          )?.userInputMessage as Record<string, unknown>
+        )?.content as string
+      ) || "";
+    const thinkingExpected = userContent.includes("<thinking_mode>enabled</thinking_mode>");
+    const transformedResponse = this.transformEventStreamToSSE(response, model, { thinkingExpected });
 
     return { response: transformedResponse, url, headers, transformedBody };
   }
 
   /**
-   * Transform AWS EventStream binary response to SSE text stream
-   * Using TransformStream instead of ReadableStream.pull() to avoid Workers timeout
+   * Transform AWS EventStream binary response to SSE text stream.
+   * Using TransformStream instead of ReadableStream.pull() to avoid Workers timeout.
+   *
+   * @param response        Upstream raw fetch response (binary EventStream).
+   * @param model           Logical model id (kept in OpenAI chunks for clients).
+   * @param opts
+   * @param opts.thinkingExpected  When true, scan inbound
+   *   `assistantResponseEvent.content` for inline `<thinking>…</thinking>`
+   *   blocks and split them into the OpenAI `delta.reasoning_content` channel.
+   *   Required for Claude on Kiro when `<thinking_mode>enabled</thinking_mode>`
+   *   is in the system prompt, because Kiro streams reasoning inline rather
+   *   than as separate `reasoningContentEvent` frames.
    */
-  transformEventStreamToSSE(response: Response, model: string) {
+  transformEventStreamToSSE(
+    response: Response,
+    model: string,
+    opts: { thinkingExpected?: boolean } = {}
+  ) {
+    const thinkingExpected = !!opts.thinkingExpected;
     const buffer = new ByteQueue();
     let chunkIndex = 0;
     const responseId = `chatcmpl-${Date.now()}`;
@@ -364,6 +402,9 @@ export class KiroExecutor extends BaseExecutor {
       seenToolIds: new Map(),
       toolArgsEmitted: new Map(),
       toolArgsBuffered: new Map(),
+      hasReasoningContent: false,
+      reasoningChunkCount: 0,
+      thinking: thinkingExpected ? { thinkingMode: false, pendingTag: "" } : undefined,
     };
 
     const transformStream = new TransformStream(
@@ -434,21 +475,75 @@ export class KiroExecutor extends BaseExecutor {
               }
               state.totalContentLength += content.length;
 
-              const chunk: JsonRecord = {
-                id: responseId,
-                object: "chat.completion.chunk",
-                created,
-                model,
-                choices: [
-                  {
-                    index: 0,
-                    delta: chunkIndex === 0 ? { role: "assistant", content } : { content },
-                    finish_reason: null,
+              if (thinkingExpected && state.thinking) {
+                // Claude on Kiro emits reasoning inline as `<thinking>…</thinking>`
+                // when `<thinking_mode>enabled</thinking_mode>` is in the system prompt.
+                // Split it into the OpenAI `reasoning_content` channel so downstream
+                // consumers see the same shape they would get from a native reasoning model.
+                const thinkingState = state.thinking;
+                splitInlineThinking(
+                  thinkingState,
+                  content,
+                  (text) => {
+                    if (!text) return;
+                    const chunk: JsonRecord = {
+                      id: responseId,
+                      object: "chat.completion.chunk",
+                      created,
+                      model,
+                      choices: [
+                        {
+                          index: 0,
+                          delta: chunkIndex === 0 ? { role: "assistant", content: text } : { content: text },
+                          finish_reason: null,
+                        },
+                      ],
+                    };
+                    chunkIndex++;
+                    controller.enqueue(TEXT_ENCODER.encode(`data: ${JSON.stringify(chunk)}\n\n`));
                   },
-                ],
-              };
-              chunkIndex++;
-              controller.enqueue(TEXT_ENCODER.encode(`data: ${JSON.stringify(chunk)}\n\n`));
+                  (reasoning) => {
+                    if (!reasoning) return;
+                    state.hasReasoningContent = true;
+                    const reasoningDelta: JsonRecord =
+                      (state.reasoningChunkCount ?? 0) === 0 && chunkIndex === 0
+                        ? { role: "assistant", reasoning_content: reasoning }
+                        : { reasoning_content: reasoning };
+                    const chunk: JsonRecord = {
+                      id: responseId,
+                      object: "chat.completion.chunk",
+                      created,
+                      model,
+                      choices: [
+                        {
+                          index: 0,
+                          delta: reasoningDelta,
+                          finish_reason: null,
+                        },
+                      ],
+                    };
+                    chunkIndex++;
+                    state.reasoningChunkCount = (state.reasoningChunkCount ?? 0) + 1;
+                    controller.enqueue(TEXT_ENCODER.encode(`data: ${JSON.stringify(chunk)}\n\n`));
+                  }
+                );
+              } else {
+                const chunk: JsonRecord = {
+                  id: responseId,
+                  object: "chat.completion.chunk",
+                  created,
+                  model,
+                  choices: [
+                    {
+                      index: 0,
+                      delta: chunkIndex === 0 ? { role: "assistant", content } : { content },
+                      finish_reason: null,
+                    },
+                  ],
+                };
+                chunkIndex++;
+                controller.enqueue(TEXT_ENCODER.encode(`data: ${JSON.stringify(chunk)}\n\n`));
+              }
             }
 
             // Handle codeEvent
@@ -644,6 +739,41 @@ export class KiroExecutor extends BaseExecutor {
           // idempotent against toolArgsEmitted if messageStopEvent already flushed them.
           flushBufferedToolArgs(state, controller, { responseId, created, model });
 
+          // Drain any pending inline-thinking tag fragment so we don't drop
+          // trailing characters when the stream ends mid-tag (e.g. `<thi`).
+          if (thinkingExpected && state.thinking) {
+            const thinkingState = state.thinking;
+            flushPendingThinking(
+              thinkingState,
+              (text) => {
+                if (!text) return;
+                const chunk: JsonRecord = {
+                  id: responseId,
+                  object: "chat.completion.chunk",
+                  created,
+                  model,
+                  choices: [{ index: 0, delta: { content: text }, finish_reason: null }],
+                };
+                chunkIndex++;
+                controller.enqueue(TEXT_ENCODER.encode(`data: ${JSON.stringify(chunk)}\n\n`));
+              },
+              (reasoning) => {
+                if (!reasoning) return;
+                const chunk: JsonRecord = {
+                  id: responseId,
+                  object: "chat.completion.chunk",
+                  created,
+                  model,
+                  choices: [
+                    { index: 0, delta: { reasoning_content: reasoning }, finish_reason: null },
+                  ],
+                };
+                chunkIndex++;
+                controller.enqueue(TEXT_ENCODER.encode(`data: ${JSON.stringify(chunk)}\n\n`));
+              }
+            );
+          }
+
           // Emit finish chunk if not already sent
           if (!state.finishEmitted) {
             state.finishEmitted = true;

open-sse/executors/kiroThinking.ts

@@ -0,0 +1,116 @@
+/**
+ * Inline `<thinking>` splitter for Claude on Kiro.
+ *
+ * Background:
+ *   When `<thinking_mode>enabled</thinking_mode>` is in the system prompt,
+ *   Claude on Kiro emits its reasoning **inline** as `<thinking>…</thinking>`
+ *   blocks inside `assistantResponseEvent.content`, rather than as separate
+ *   `reasoningContentEvent` frames. To match the OpenAI streaming shape that
+ *   downstream translators (Anthropic /thinking_blocks, Claude SSE, etc.)
+ *   expect, we split that inline reasoning back out and route it to the
+ *   `delta.reasoning_content` channel instead of `delta.content`.
+ *
+ * The implementation is split into pure functions so it can be unit-tested
+ * without dragging in the rest of the executor stack (proxy-agent, AWS
+ * EventStream parser, etc.). The KiroExecutor wires these helpers into its
+ * TransformStream by passing controller-bound emit callbacks.
+ *
+ * Ported from decolua/9router#1273 (kiroThinking.js) by Amin Fathullah.
+ */
+
+/** Mutable state carried across `splitInlineThinking` calls. */
+export type KiroThinkingState = {
+  /** True while the cursor is inside a `<thinking>` block. */
+  thinkingMode: boolean;
+  /**
+   * Characters held back because they might be the start of a tag we'll
+   * complete on the next slice (e.g. `<thi`).
+   */
+  pendingTag: string;
+};
+
+/**
+ * Stream-safe splitter. Walks one slice of upstream content at a time and
+ * routes characters to either the content channel or the reasoning channel
+ * based on the current `<thinking>` state.
+ *
+ * State is mutated on `state` so a tag split between frames (e.g. `…</think`
+ * followed by `ing>foo`) is still recognised.
+ *
+ * @param state   Mutable state carried across calls. Initialise with
+ *                `{ thinkingMode: false, pendingTag: "" }`.
+ * @param raw     Next slice from `assistantResponseEvent.content`. May be empty
+ *                or null/undefined (no-op).
+ * @param onContent   Called with text that should land in `delta.content`.
+ * @param onReasoning Called with text that should land in `delta.reasoning_content`.
+ */
+export function splitInlineThinking(
+  state: KiroThinkingState,
+  raw: string | null | undefined,
+  onContent: (s: string) => void,
+  onReasoning: (s: string) => void
+): void {
+  let text = (state.pendingTag || "") + (raw || "");
+  state.pendingTag = "";
+
+  // Maximum length of an unfinished tag we might still complete on the next
+  // frame: `</thinking>` is the longest at 11 chars.
+  const PARTIAL_MAX = 11;
+
+  while (text.length > 0) {
+    const target = state.thinkingMode ? "</thinking>" : "<thinking>";
+    const idx = text.indexOf(target);
+
+    if (idx === -1) {
+      // No full target tag in `text`. Look for a possible partial at the end
+      // so we can complete it on the next frame.
+      let holdFrom = text.length;
+      for (let i = Math.max(0, text.length - PARTIAL_MAX); i < text.length; i++) {
+        const tail = text.slice(i);
+        if (target.startsWith(tail) && tail.length > 0) {
+          holdFrom = i;
+          break;
+        }
+      }
+      const flushable = text.slice(0, holdFrom);
+      if (flushable) {
+        if (state.thinkingMode) onReasoning(flushable);
+        else onContent(flushable);
+      }
+      state.pendingTag = text.slice(holdFrom);
+      return;
+    }
+
+    // Found a complete target tag. Flush everything before it in the current
+    // mode, flip the mode, and keep walking the remainder.
+    const before = text.slice(0, idx);
+    if (before) {
+      if (state.thinkingMode) onReasoning(before);
+      else onContent(before);
+    }
+    state.thinkingMode = !state.thinkingMode;
+    text = text.slice(idx + target.length);
+  }
+}
+
+/**
+ * Drain whatever is left in `state.pendingTag` at end-of-stream. Routes the
+ * leftover characters to whichever channel matches the current
+ * `state.thinkingMode` so we don't silently lose data when the stream ends
+ * mid-tag (e.g. `<thi`).
+ *
+ * @param state       Mutable state shared with `splitInlineThinking`.
+ * @param onContent   Called with text that should land in `delta.content`.
+ * @param onReasoning Called with text that should land in `delta.reasoning_content`.
+ */
+export function flushPendingThinking(
+  state: KiroThinkingState,
+  onContent: (s: string) => void,
+  onReasoning: (s: string) => void
+): void {
+  if (!state.pendingTag) return;
+  const leftover = state.pendingTag;
+  state.pendingTag = "";
+  if (state.thinkingMode) onReasoning(leftover);
+  else onContent(leftover);
+}

tests/unit/kiroThinking.test.ts

@@ -0,0 +1,171 @@
+/**
+ * Unit tests for the Kiro inline `<thinking>` stream splitter.
+ *
+ * Ported from decolua/9router#1273 (tests/unit/kiroThinking.test.js).
+ * Tests are framework-agnostic: Node.js native test runner (tsx/esm).
+ */
+import { describe, it } from "node:test";
+import assert from "node:assert/strict";
+import { splitInlineThinking, flushPendingThinking } from "../../open-sse/executors/kiroThinking.ts";
+
+/** Build a fresh state + recorders for each test. */
+function makeHarness() {
+  const state = { thinkingMode: false, pendingTag: "" };
+  const contentParts: string[] = [];
+  const reasoningParts: string[] = [];
+  const onContent = (s: string) => contentParts.push(s);
+  const onReasoning = (s: string) => reasoningParts.push(s);
+  const feed = (raw: string | null | undefined) =>
+    splitInlineThinking(state, raw, onContent, onReasoning);
+  const flush = () => flushPendingThinking(state, onContent, onReasoning);
+  return {
+    state,
+    feed,
+    flush,
+    get content() {
+      return contentParts.join("");
+    },
+    get reasoning() {
+      return reasoningParts.join("");
+    },
+  };
+}
+
+describe("splitInlineThinking", () => {
+  it("passes through plain content with no tags", () => {
+    const h = makeHarness();
+    h.feed("Bonjour, this is just an answer.");
+    h.flush();
+    assert.equal(h.content, "Bonjour, this is just an answer.");
+    assert.equal(h.reasoning, "");
+    assert.equal(h.state.thinkingMode, false);
+    assert.equal(h.state.pendingTag, "");
+  });
+
+  it("splits a single-shot input with one full block", () => {
+    const h = makeHarness();
+    h.feed("Hello <thinking>secret thoughts</thinking> world");
+    h.flush();
+    assert.equal(h.content, "Hello  world");
+    assert.equal(h.reasoning, "secret thoughts");
+    assert.equal(h.state.thinkingMode, false);
+  });
+
+  it("handles a tag split across two slices (open tag)", () => {
+    // The opening `<thinking>` arrives across two reads.
+    const h = makeHarness();
+    h.feed("Hi <thi");
+    assert.equal(h.content, "Hi "); // only the safe prefix is flushed
+    assert.equal(h.state.pendingTag, "<thi");
+
+    h.feed("nking>thoughts</thinking> bye");
+    h.flush();
+    assert.equal(h.content, "Hi  bye");
+    assert.equal(h.reasoning, "thoughts");
+    assert.equal(h.state.pendingTag, "");
+  });
+
+  it("handles a tag split across two slices (close tag)", () => {
+    const h = makeHarness();
+    h.feed("<thinking>step 1</thi");
+    assert.equal(h.reasoning, "step 1");
+    assert.equal(h.state.thinkingMode, true);
+    assert.equal(h.state.pendingTag, "</thi");
+
+    h.feed("nking>final answer");
+    h.flush();
+    assert.equal(h.content, "final answer");
+    assert.equal(h.reasoning, "step 1");
+    assert.equal(h.state.thinkingMode, false);
+  });
+
+  it("handles a tag split character-by-character", () => {
+    const h = makeHarness();
+    const stream = "before <thinking>hidden</thinking>after";
+    for (const ch of stream) h.feed(ch);
+    h.flush();
+    assert.equal(h.content, "before after");
+    assert.equal(h.reasoning, "hidden");
+    assert.equal(h.state.thinkingMode, false);
+    assert.equal(h.state.pendingTag, "");
+  });
+
+  it("handles multiple thinking blocks in sequence", () => {
+    const h = makeHarness();
+    h.feed("<thinking>plan A</thinking>step1<thinking>plan B</thinking>step2");
+    h.flush();
+    assert.equal(h.content, "step1step2");
+    assert.equal(h.reasoning, "plan Aplan B");
+    assert.equal(h.state.thinkingMode, false);
+  });
+
+  it("emits leftover content when the stream ends mid-tag", () => {
+    const h = makeHarness();
+    h.feed("ok <thi");
+    assert.equal(h.state.pendingTag, "<thi");
+    h.flush();
+    // Not in thinking mode — leftover goes to content.
+    assert.equal(h.content, "ok <thi");
+    assert.equal(h.reasoning, "");
+  });
+
+  it("emits leftover reasoning when the stream ends mid-close-tag", () => {
+    const h = makeHarness();
+    h.feed("<thinking>partial</thi");
+    assert.equal(h.state.thinkingMode, true);
+    assert.equal(h.state.pendingTag, "</thi");
+    h.flush();
+    // Leftover stays in reasoning because we never saw the close tag.
+    assert.equal(h.reasoning, "partial</thi");
+    assert.equal(h.content, "");
+  });
+
+  it("does not consume tag-shaped content that isn't a real tag boundary", () => {
+    const h = makeHarness();
+    h.feed("a<b>c</b>d");
+    h.flush();
+    assert.equal(h.content, "a<b>c</b>d");
+    assert.equal(h.reasoning, "");
+  });
+
+  it("only holds back trailing characters that look like a partial OPEN tag (outside thinking mode)", () => {
+    const h = makeHarness();
+    h.feed("answer text <think");
+    assert.equal(h.content, "answer text "); // safe prefix flushed eagerly
+    assert.equal(h.state.pendingTag, "<think");
+
+    h.feed("ing>secret</thinking>tail");
+    h.flush();
+    assert.equal(h.content, "answer text tail");
+    assert.equal(h.reasoning, "secret");
+  });
+
+  it("does not hold back partial CLOSE tag while outside thinking mode", () => {
+    const h = makeHarness();
+    h.feed("answer text </think");
+    h.feed(" something else");
+    h.flush();
+    assert.equal(h.content, "answer text </think something else");
+    assert.equal(h.reasoning, "");
+  });
+
+  it("survives empty / null / undefined slices", () => {
+    const h = makeHarness();
+    h.feed("");
+    h.feed(undefined);
+    h.feed(null);
+    h.feed("hello");
+    h.flush();
+    assert.equal(h.content, "hello");
+    assert.equal(h.reasoning, "");
+  });
+
+  it("flushPendingThinking is a no-op when nothing is pending", () => {
+    const h = makeHarness();
+    h.feed("just content");
+    h.flush();
+    h.flush(); // second flush should be a no-op
+    assert.equal(h.content, "just content");
+    assert.equal(h.reasoning, "");
+  });
+});