From 61d7566ca1b1fb53bf1688f6e400c61cf2eb8f4a Mon Sep 17 00:00:00 2001 From: Mikhail Salnikov <14850941+mikhailsal@users.noreply.github.com> Date: Sun, 29 Mar 2026 20:51:26 +0300 Subject: [PATCH 1/4] fix(stream): normalize delta.reasoning to reasoning_content in SSE streaming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NVIDIA kimi-k2.5 (and potentially other providers) send reasoning tokens as `delta.reasoning` in SSE streaming chunks instead of the standard OpenAI `delta.reasoning_content` field. This caused reasoning content to be silently dropped during stream passthrough — clients received only the final answer with no reasoning separation. The non-streaming sanitizer (responseSanitizer.ts) already handled this alias, but the streaming pipeline did not. Fix applied in 4 locations: - stream.ts passthrough: normalize + force re-serialize sanitized chunk - stream.ts translate: accumulate reasoning from delta.reasoning - sseParser.ts: collect delta.reasoning in parseSSEToOpenAIResponse - streamPayloadCollector.ts: collect delta.reasoning in buildOpenAISummary --- open-sse/handlers/sseParser.ts | 4 ++++ open-sse/utils/stream.ts | 24 ++++++++++++++++++++++++ open-sse/utils/streamPayloadCollector.ts | 4 ++++ 3 files changed, 32 insertions(+) diff --git a/open-sse/handlers/sseParser.ts b/open-sse/handlers/sseParser.ts index e1e61361..4a9626ac 100644 --- a/open-sse/handlers/sseParser.ts +++ b/open-sse/handlers/sseParser.ts @@ -52,6 +52,10 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) { if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) { reasoningParts.push(delta.reasoning_content); } + // Normalize `reasoning` alias (NVIDIA kimi-k2.5 etc.) + if (typeof delta.reasoning === "string" && delta.reasoning.length > 0 && !delta.reasoning_content) { + reasoningParts.push(delta.reasoning); + } // T18: Accumulate tool calls correctly across streamed chunks if (delta.tool_calls) { diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 4c30fe37..6c274663 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -314,6 +314,12 @@ export function createSSEStream(options: StreamOptions = {}) { const delta = parsed.choices?.[0]?.delta; + // Normalize `reasoning` alias → `reasoning_content` (NVIDIA kimi-k2.5 etc.) + if (delta?.reasoning && typeof delta.reasoning === "string" && !delta.reasoning_content) { + delta.reasoning_content = delta.reasoning; + delete delta.reasoning; + } + // Extract tags from streaming content if (delta?.content && typeof delta.content === "string") { const { content, thinking } = extractThinkingFromContent(delta.content); @@ -323,6 +329,14 @@ export function createSSEStream(options: StreamOptions = {}) { } } + // If reasoning was normalized (reasoning → reasoning_content) or + // tags were extracted, force re-serialization so the client sees + // the standard `reasoning_content` field instead of the raw provider line. + if (delta?.reasoning_content && !injectedUsage) { + output = `data: ${JSON.stringify(parsed)}\n`; + injectedUsage = true; + } + // T18: Track if we saw tool calls & accumulate for call log if (delta?.tool_calls && delta.tool_calls.length > 0) { passthroughHasToolCalls = true; @@ -483,6 +497,16 @@ export function createSSEStream(options: StreamOptions = {}) { if (state?.accumulatedContent !== undefined) state.accumulatedContent += r; } } + // Normalize `reasoning` alias → `reasoning_content` (NVIDIA kimi-k2.5 etc.) + if (parsed.choices?.[0]?.delta?.reasoning && !parsed.choices?.[0]?.delta?.reasoning_content) { + const r = parsed.choices[0].delta.reasoning; + if (typeof r === "string") { + parsed.choices[0].delta.reasoning_content = r; + delete parsed.choices[0].delta.reasoning; + totalContentLength += r.length; + if (state?.accumulatedContent !== undefined) state.accumulatedContent += r; + } + } // Gemini format - may have multiple parts if (parsed.candidates?.[0]?.content?.parts) { diff --git a/open-sse/utils/streamPayloadCollector.ts b/open-sse/utils/streamPayloadCollector.ts index 961dfc46..427770fa 100644 --- a/open-sse/utils/streamPayloadCollector.ts +++ b/open-sse/utils/streamPayloadCollector.ts @@ -157,6 +157,10 @@ function buildOpenAISummary(events: StructuredSSEEvent[], fallbackModel?: string if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) { reasoningParts.push(delta.reasoning_content); } + // Normalize `reasoning` alias (NVIDIA kimi-k2.5 etc.) + if (typeof delta.reasoning === "string" && delta.reasoning.length > 0 && !delta.reasoning_content) { + reasoningParts.push(delta.reasoning); + } if (Array.isArray(delta.tool_calls)) { for (const item of delta.tool_calls) { From 4083447c3f418745dac24368362a68fc060ff890 Mon Sep 17 00:00:00 2001 From: Mikhail Salnikov <14850941+mikhailsal@users.noreply.github.com> Date: Sun, 29 Mar 2026 21:18:46 +0300 Subject: [PATCH 2/4] fix: eliminate injectedUsage reuse bug and add reasoning alias tests - Detect delta.reasoning alias before sanitizeStreamingChunk() which already normalizes it, removing dead post-sanitization normalization - Replace injectedUsage reuse with separate needsReserialization flag so reasoning re-serialization cannot block finish_reason/usage mutations on the same SSE chunk (fixes CRITICAL review finding) - Add unit test for parseSSEToOpenAIResponse reasoning alias - Add unit test for buildStreamSummaryFromEvents reasoning alias --- open-sse/utils/stream.ts | 32 ++++++++-------- tests/unit/plan3-p0.test.mjs | 26 +++++++++++++ tests/unit/request-log-payloads.test.mjs | 48 ++++++++++++++++++++++++ 3 files changed, 91 insertions(+), 15 deletions(-) diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 6c274663..bcf8eedb 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -304,6 +304,14 @@ export function createSSEStream(options: StreamOptions = {}) { } } else { // Chat Completions: full sanitization pipeline + + // Detect reasoning alias before sanitization strips it + const hadReasoningAlias = !!( + parsed.choices?.[0]?.delta?.reasoning && + typeof parsed.choices[0].delta.reasoning === "string" && + !parsed.choices[0].delta.reasoning_content + ); + parsed = sanitizeStreamingChunk(parsed); const idFixed = fixInvalidId(parsed); @@ -314,12 +322,6 @@ export function createSSEStream(options: StreamOptions = {}) { const delta = parsed.choices?.[0]?.delta; - // Normalize `reasoning` alias → `reasoning_content` (NVIDIA kimi-k2.5 etc.) - if (delta?.reasoning && typeof delta.reasoning === "string" && !delta.reasoning_content) { - delta.reasoning_content = delta.reasoning; - delete delta.reasoning; - } - // Extract tags from streaming content if (delta?.content && typeof delta.content === "string") { const { content, thinking } = extractThinkingFromContent(delta.content); @@ -329,13 +331,10 @@ export function createSSEStream(options: StreamOptions = {}) { } } - // If reasoning was normalized (reasoning → reasoning_content) or - // tags were extracted, force re-serialization so the client sees - // the standard `reasoning_content` field instead of the raw provider line. - if (delta?.reasoning_content && !injectedUsage) { - output = `data: ${JSON.stringify(parsed)}\n`; - injectedUsage = true; - } + // Track whether we need to re-serialize (separate from injectedUsage + // to avoid blocking subsequent finish_reason / usage mutations) + const needsReserialization = + hadReasoningAlias || (delta?.content === "" && delta?.reasoning_content); // T18: Track if we saw tool calls & accumulate for call log if (delta?.tool_calls && delta.tool_calls.length > 0) { @@ -412,7 +411,7 @@ export function createSSEStream(options: StreamOptions = {}) { parsed.usage = filterUsageForFormat(buffered, FORMATS.OPENAI); output = `data: ${JSON.stringify(parsed)}\n`; injectedUsage = true; - } else if (idFixed) { + } else if (idFixed || needsReserialization) { output = `data: ${JSON.stringify(parsed)}\n`; injectedUsage = true; } @@ -498,7 +497,10 @@ export function createSSEStream(options: StreamOptions = {}) { } } // Normalize `reasoning` alias → `reasoning_content` (NVIDIA kimi-k2.5 etc.) - if (parsed.choices?.[0]?.delta?.reasoning && !parsed.choices?.[0]?.delta?.reasoning_content) { + if ( + parsed.choices?.[0]?.delta?.reasoning && + !parsed.choices?.[0]?.delta?.reasoning_content + ) { const r = parsed.choices[0].delta.reasoning; if (typeof r === "string") { parsed.choices[0].delta.reasoning_content = r; diff --git a/tests/unit/plan3-p0.test.mjs b/tests/unit/plan3-p0.test.mjs index 6fa213b9..b59cd482 100644 --- a/tests/unit/plan3-p0.test.mjs +++ b/tests/unit/plan3-p0.test.mjs @@ -503,3 +503,29 @@ test("parseSSEToOpenAIResponse merges split tool call chunks by id without dupli assert.equal(parsed.choices[0].message.tool_calls[0].function.name, "sum"); assert.equal(parsed.choices[0].message.tool_calls[0].function.arguments, '{"a":1}'); }); + +test("parseSSEToOpenAIResponse normalizes delta.reasoning alias to reasoning_content", () => { + const rawSSE = [ + `data: ${JSON.stringify({ + id: "chatcmpl_2", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { reasoning: "Let me think..." } }], + })}`, + `data: ${JSON.stringify({ + id: "chatcmpl_2", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { reasoning: " The answer is 4." } }], + })}`, + `data: ${JSON.stringify({ + id: "chatcmpl_2", + object: "chat.completion.chunk", + choices: [{ index: 0, delta: { content: "2+2=4" }, finish_reason: "stop" }], + })}`, + "data: [DONE]", + ].join("\n"); + + const parsed = parseSSEToOpenAIResponse(rawSSE, "moonshotai/kimi-k2.5"); + assert.ok(parsed); + assert.equal(parsed.choices[0].message.reasoning_content, "Let me think... The answer is 4."); + assert.equal(parsed.choices[0].message.content, "2+2=4"); +}); diff --git a/tests/unit/request-log-payloads.test.mjs b/tests/unit/request-log-payloads.test.mjs index dc6ae405..6a4ece1e 100644 --- a/tests/unit/request-log-payloads.test.mjs +++ b/tests/unit/request-log-payloads.test.mjs @@ -155,3 +155,51 @@ test("builds compact Claude stream summary for detailed logs", () => { assert.equal(compact.usage.output_tokens, 7); assert.equal(compact._omniroute_stream.eventCount, 4); }); + +test("builds compact OpenAI summary with reasoning alias (delta.reasoning)", () => { + const collector = createStructuredSSECollector({ stage: "provider_response" }); + + collector.push({ + id: "chatcmpl_r1", + object: "chat.completion.chunk", + created: 100, + model: "moonshotai/kimi-k2.5", + choices: [{ index: 0, delta: { role: "assistant" } }], + }); + collector.push({ + id: "chatcmpl_r1", + object: "chat.completion.chunk", + created: 100, + model: "moonshotai/kimi-k2.5", + choices: [{ index: 0, delta: { reasoning: "Let me think..." } }], + }); + collector.push({ + id: "chatcmpl_r1", + object: "chat.completion.chunk", + created: 100, + model: "moonshotai/kimi-k2.5", + choices: [{ index: 0, delta: { content: "The answer is 4." } }], + }); + collector.push({ + id: "chatcmpl_r1", + object: "chat.completion.chunk", + created: 100, + model: "moonshotai/kimi-k2.5", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + usage: { prompt_tokens: 10, completion_tokens: 20, total_tokens: 30 }, + }); + + const summary = buildStreamSummaryFromEvents( + collector.getEvents(), + FORMATS.OPENAI, + "moonshotai/kimi-k2.5" + ); + const compact = compactStructuredStreamPayload( + collector.build(summary, { includeEvents: false }) + ); + + assert.equal(compact.object, "chat.completion"); + assert.equal(compact.choices[0].message.content, "The answer is 4."); + assert.equal(compact.choices[0].message.reasoning_content, "Let me think..."); + assert.equal(compact.choices[0].finish_reason, "stop"); +}); From aa93276e6e87fa2f7720f34f766c21bd36fc1d47 Mon Sep 17 00:00:00 2001 From: Mikhail Salnikov <14850941+mikhailsal@users.noreply.github.com> Date: Sun, 29 Mar 2026 21:41:32 +0300 Subject: [PATCH 3/4] fix(stream): separate reasoning from content in passthrough response body The passthroughAccumulatedContent variable was mixing delta.content and delta.reasoning_content into one string, causing the client_response log and responseBody to lose reasoning separation. - Add passthroughAccumulatedReasoning accumulator for reasoning deltas - Set message.reasoning_content in responseBody when reasoning exists - Only accumulate delta.content into passthroughAccumulatedContent --- open-sse/utils/stream.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index bcf8eedb..69c8ee95 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -159,8 +159,9 @@ export function createSSEStream(options: StreamOptions = {}) { // Track content length for usage estimation (both modes) let totalContentLength = 0; - // Passthrough: accumulate content for call log response body + // Passthrough: accumulate content and reasoning separately for call log response body let passthroughAccumulatedContent = ""; + let passthroughAccumulatedReasoning = ""; // Guard against duplicate [DONE] events — ensures exactly one per stream let doneSent = false; @@ -378,7 +379,7 @@ export function createSSEStream(options: StreamOptions = {}) { if (typeof delta?.content === "string") passthroughAccumulatedContent += delta.content; if (typeof delta?.reasoning_content === "string") - passthroughAccumulatedContent += delta.reasoning_content; + passthroughAccumulatedReasoning += delta.reasoning_content; const extracted = extractUsage(parsed); if (extracted) { @@ -661,6 +662,10 @@ export function createSSEStream(options: StreamOptions = {}) { role: "assistant", content: content || null, }; + const reasoning = passthroughAccumulatedReasoning.trim(); + if (reasoning) { + message.reasoning_content = reasoning; + } if (passthroughToolCalls.size > 0) { message.tool_calls = [...passthroughToolCalls.values()].sort( (a, b) => a.index - b.index From d3a24446b84da59aeb55987ed3fb874b61c32b03 Mon Sep 17 00:00:00 2001 From: Mikhail Salnikov <14850941+mikhailsal@users.noreply.github.com> Date: Sun, 29 Mar 2026 21:44:13 +0300 Subject: [PATCH 4/4] fix: trim leading whitespace from assembled content in log summaries NVIDIA and other providers emit token deltas with leading spaces (e.g. ' The', ' user'). When joined, these produce a leading space in the provider_response and parsed non-streaming response logs. Trim the joined content and reasoning_content in both buildOpenAISummary and parseSSEToOpenAIResponse for consistent log output. --- open-sse/handlers/sseParser.ts | 8 +++++--- open-sse/utils/streamPayloadCollector.ts | 8 +++++--- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/open-sse/handlers/sseParser.ts b/open-sse/handlers/sseParser.ts index 4a9626ac..aa579010 100644 --- a/open-sse/handlers/sseParser.ts +++ b/open-sse/handlers/sseParser.ts @@ -98,12 +98,14 @@ export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) { } } + const joinedContent = contentParts.length > 0 ? contentParts.join("").trim() : null; + const joinedReasoning = reasoningParts.length > 0 ? reasoningParts.join("").trim() : null; const message: Record = { role: "assistant", - content: contentParts.length > 0 ? contentParts.join("") : null, + content: joinedContent || null, }; - if (reasoningParts.length > 0) { - message.reasoning_content = reasoningParts.join(""); + if (joinedReasoning) { + message.reasoning_content = joinedReasoning; } const finalToolCalls = [...accumulatedToolCalls.values()].filter(Boolean).sort((a, b) => { diff --git a/open-sse/utils/streamPayloadCollector.ts b/open-sse/utils/streamPayloadCollector.ts index 427770fa..58a20cbb 100644 --- a/open-sse/utils/streamPayloadCollector.ts +++ b/open-sse/utils/streamPayloadCollector.ts @@ -207,12 +207,14 @@ function buildOpenAISummary(events: StructuredSSEEvent[], fallbackModel?: string } } + const joinedContent = contentParts.length > 0 ? contentParts.join("").trim() : null; + const joinedReasoning = reasoningParts.length > 0 ? reasoningParts.join("").trim() : null; const message: JsonRecord = { role: "assistant", - content: contentParts.length > 0 ? contentParts.join("") : null, + content: joinedContent || null, }; - if (reasoningParts.length > 0) { - message.reasoning_content = reasoningParts.join(""); + if (joinedReasoning) { + message.reasoning_content = joinedReasoning; } const finalToolCalls = [...toolCalls.values()].sort((a, b) => a.index - b.index);