From 2dbd3e933aa61af4b5c0b74c4a25e2a1da400cba Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Fri, 26 Jun 2026 10:56:54 -0300 Subject: [PATCH] feat(sse): Kiro inline stream splitter (#4911) Integrated into release/v3.8.38 (leva 5) --- CHANGELOG.md | 1 + open-sse/executors/kiro.ts | 170 ++++++++++++++++++++++++---- open-sse/executors/kiroThinking.ts | 116 +++++++++++++++++++ tests/unit/kiroThinking.test.ts | 171 +++++++++++++++++++++++++++++ 4 files changed, 438 insertions(+), 20 deletions(-) create mode 100644 open-sse/executors/kiroThinking.ts create mode 100644 tests/unit/kiroThinking.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e864c33b3..3be0e6c510 100644 --- a/CHANGELOG.md +++ b/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 `` stream splitter — when `enabled` 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 diff --git a/open-sse/executors/kiro.ts b/open-sse/executors/kiro.ts index 382b11919a..614efbaf3b 100644 --- a/open-sse/executors/kiro.ts +++ b/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; @@ -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 `…` 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; + const userContent = + ( + ( + ( + (tb?.conversationState as Record) + ?.currentMessage as Record + )?.userInputMessage as Record + )?.content as string + ) || ""; + const thinkingExpected = userContent.includes("enabled"); + 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 `…` + * blocks and split them into the OpenAI `delta.reasoning_content` channel. + * Required for Claude on Kiro when `enabled` + * 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 `…` + // when `enabled` 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. ` { + 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; diff --git a/open-sse/executors/kiroThinking.ts b/open-sse/executors/kiroThinking.ts new file mode 100644 index 0000000000..49ca3e324b --- /dev/null +++ b/open-sse/executors/kiroThinking.ts @@ -0,0 +1,116 @@ +/** + * Inline `` splitter for Claude on Kiro. + * + * Background: + * When `enabled` is in the system prompt, + * Claude on Kiro emits its reasoning **inline** as `…` + * 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 `` block. */ + thinkingMode: boolean; + /** + * Characters held back because they might be the start of a tag we'll + * complete on the next slice (e.g. `` state. + * + * State is mutated on `state` so a tag split between frames (e.g. `…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: `` is the longest at 11 chars. + const PARTIAL_MAX = 11; + + while (text.length > 0) { + const target = state.thinkingMode ? "" : ""; + 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. ` void, + onReasoning: (s: string) => void +): void { + if (!state.pendingTag) return; + const leftover = state.pendingTag; + state.pendingTag = ""; + if (state.thinkingMode) onReasoning(leftover); + else onContent(leftover); +} diff --git a/tests/unit/kiroThinking.test.ts b/tests/unit/kiroThinking.test.ts new file mode 100644 index 0000000000..c4da9a5ce9 --- /dev/null +++ b/tests/unit/kiroThinking.test.ts @@ -0,0 +1,171 @@ +/** + * Unit tests for the Kiro inline `` 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 secret thoughts 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 `` arrives across two reads. + const h = makeHarness(); + h.feed("Hi thoughts 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("step 1final 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 hiddenafter"; + 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("plan Astep1plan Bstep2"); + 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 { + const h = makeHarness(); + h.feed("partial { + const h = makeHarness(); + h.feed("acd"); + h.flush(); + assert.equal(h.content, "acd"); + 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 secrettail"); + 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 { + 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, ""); + }); +});