From 9d673e5f9e24fd341f2bf4cf2aa42a8bf100ca2c Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Thu, 25 Jun 2026 23:34:23 -0300 Subject: [PATCH] fix(sse): close reasoning before message content in Responses stream (#4848) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.37 — cherry-picked defining commit onto release tip; CHANGELOG re-merged; tests green. --- open-sse/transformer/responsesTransformer.ts | 47 ++++-- .../translator/response/openai-responses.ts | 20 ++- ...reasoning-close-before-message-466.test.ts | 158 ++++++++++++++++++ 3 files changed, 209 insertions(+), 16 deletions(-) create mode 100644 tests/unit/responses-reasoning-close-before-message-466.test.ts diff --git a/open-sse/transformer/responsesTransformer.ts b/open-sse/transformer/responsesTransformer.ts index 6eebf965e0..d7ff92d6e7 100644 --- a/open-sse/transformer/responsesTransformer.ts +++ b/open-sse/transformer/responsesTransformer.ts @@ -416,6 +416,14 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv // Handle text content (may contain tags) if (delta.content) { + // Close reasoning if it was opened via native reasoning_content + // and is still open, before emitting message content. Without this + // the reasoning item is never closed and the message reuses the + // reasoning output_index, producing a protocol-invalid stream. + if (state.reasoningId && !state.reasoningDone) { + closeReasoning(controller); + } + let content = delta.content; if (content.includes("")) { @@ -442,31 +450,36 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv // Regular text content if (content) { + // Use a distinct output_index for the message when reasoning was + // emitted, so the message item does not collide with the + // reasoning item's output_index. + const msgIdx = state.reasoningId ? state.reasoningIndex + 1 : idx; + // Fix for #1211: Strip leading double-newlines / blank spaces from the very first text chunk - if (!state.msgTextBuf[idx]) { + if (!state.msgTextBuf[msgIdx]) { content = content.trimStart(); } if (!content) continue; - if (!state.msgItemAdded[idx]) { - state.msgItemAdded[idx] = true; - const msgId = `msg_${state.responseId}_${idx}`; + if (!state.msgItemAdded[msgIdx]) { + state.msgItemAdded[msgIdx] = true; + const msgId = `msg_${state.responseId}_${msgIdx}`; emit(controller, "response.output_item.added", { type: "response.output_item.added", - output_index: idx, + output_index: msgIdx, item: { id: msgId, type: "message", content: [], role: "assistant" }, }); } - if (!state.msgContentAdded[idx]) { - state.msgContentAdded[idx] = true; + if (!state.msgContentAdded[msgIdx]) { + state.msgContentAdded[msgIdx] = true; emit(controller, "response.content_part.added", { type: "response.content_part.added", - item_id: `msg_${state.responseId}_${idx}`, - output_index: idx, + item_id: `msg_${state.responseId}_${msgIdx}`, + output_index: msgIdx, content_index: 0, part: { type: "output_text", annotations: [], logprobs: [], text: "" }, }); @@ -474,21 +487,27 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv emit(controller, "response.output_text.delta", { type: "response.output_text.delta", - item_id: `msg_${state.responseId}_${idx}`, - output_index: idx, + item_id: `msg_${state.responseId}_${msgIdx}`, + output_index: msgIdx, content_index: 0, delta: content, logprobs: [], }); - if (!state.msgTextBuf[idx]) state.msgTextBuf[idx] = ""; - state.msgTextBuf[idx] += content; + if (!state.msgTextBuf[msgIdx]) state.msgTextBuf[msgIdx] = ""; + state.msgTextBuf[msgIdx] += content; } } // Handle tool_calls if (delta.tool_calls) { - closeMessage(controller, idx); + // Close reasoning first so tool calls do not collide with an + // open reasoning item, then close the message at its real index. + if (state.reasoningId && !state.reasoningDone) { + closeReasoning(controller); + } + const msgIdx = state.reasoningId ? state.reasoningIndex + 1 : idx; + closeMessage(controller, msgIdx); for (const tc of delta.tool_calls) { const tcIdx = tc.index ?? 0; diff --git a/open-sse/translator/response/openai-responses.ts b/open-sse/translator/response/openai-responses.ts index caa52ecacc..5fc77e589a 100644 --- a/open-sse/translator/response/openai-responses.ts +++ b/open-sse/translator/response/openai-responses.ts @@ -125,6 +125,13 @@ export function openaiToOpenAIResponsesResponse(chunk, state) { // Handle text content if (delta.content) { + // Close reasoning if it was opened via native reasoning_content and is + // still open, before emitting message content. Otherwise the reasoning + // item is never closed and the message reuses its output_index. + if (state.reasoningId && !state.reasoningDone) { + closeReasoning(state, emit); + } + let content = delta.content; if (content.includes("")) { @@ -149,13 +156,22 @@ export function openaiToOpenAIResponsesResponse(chunk, state) { } if (content) { - emitTextContent(state, emit, idx, content); + // Use a distinct output_index for the message when reasoning was + // emitted, so the message item does not collide with the reasoning item. + const msgIdx = state.reasoningId ? state.reasoningIndex + 1 : idx; + emitTextContent(state, emit, msgIdx, content); } } // Handle tool_calls if (delta.tool_calls) { - closeMessage(state, emit, idx); + // Close reasoning first so tool calls do not collide with an open + // reasoning item, then close the message at its real index. + if (state.reasoningId && !state.reasoningDone) { + closeReasoning(state, emit); + } + const msgIdx = state.reasoningId ? state.reasoningIndex + 1 : idx; + closeMessage(state, emit, msgIdx); for (const tc of delta.tool_calls) { emitToolCall(state, emit, tc); } diff --git a/tests/unit/responses-reasoning-close-before-message-466.test.ts b/tests/unit/responses-reasoning-close-before-message-466.test.ts new file mode 100644 index 0000000000..e277aec109 --- /dev/null +++ b/tests/unit/responses-reasoning-close-before-message-466.test.ts @@ -0,0 +1,158 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +// Regression guard for the Responses-API streaming-protocol bug where native +// reasoning (`delta.reasoning_content`, no tags) opened a reasoning +// item that was never closed before the message content, and the message +// reused the reasoning item's output_index — producing a protocol-invalid +// stream. After the fix, reasoning is closed at the top of the content/ +// tool_calls handlers and the message is routed to `reasoningIndex + 1`. + +const { createResponsesApiTransformStream } = await import( + "../../open-sse/transformer/responsesTransformer.ts" +); +const { openaiToOpenAIResponsesResponse } = await import( + "../../open-sse/translator/response/openai-responses.ts" +); +const { initState } = await import("../../open-sse/translator/index.ts"); +const { FORMATS } = await import("../../open-sse/translator/formats.ts"); + +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); + +async function runTransformStream(chunks) { + const stream = createResponsesApiTransformStream(null); + const writer = stream.writable.getWriter(); + const reader = stream.readable.getReader(); + + const output = []; + const readerTask = (async () => { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + output.push(decoder.decode(value)); + } + })(); + + for (const chunk of chunks) { + await writer.write(encoder.encode(chunk)); + } + await writer.close(); + await readerTask; + + return output.join(""); +} + +function parseSseOutput(output) { + return output + .trim() + .split("\n\n") + .map((entry) => { + const lines = entry.split("\n"); + const eventLine = lines.find((line) => line.startsWith("event: ")); + const dataLine = lines.find((line) => line.startsWith("data: ")); + return { + event: eventLine ? eventLine.slice("event: ".length) : null, + data: dataLine ? dataLine.slice("data: ".length) : null, + }; + }); +} + +test("transformer: native reasoning is closed before message content and uses a distinct output_index", async () => { + const output = await runTransformStream([ + 'data: {"id":"chatcmpl_1","choices":[{"index":0,"delta":{"reasoning_content":"thinking"}}]}\n\n', + 'data: {"choices":[{"index":0,"delta":{"content":"answer"}}]}\n\n', + 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n', + ]); + + const events = parseSseOutput(output).map((e) => { + let data = null; + if (e.data && e.data !== "[DONE]") { + try { + data = JSON.parse(e.data); + } catch { + data = null; + } + } + return { event: e.event, data }; + }); + + // The reasoning item must be closed (output_item.done with a reasoning item) + // BEFORE the first message output_item.added is emitted. + const reasoningDoneIdx = events.findIndex( + (e) => e.event === "response.output_item.done" && e.data?.item?.type === "reasoning" + ); + const messageAddedIdx = events.findIndex( + (e) => e.event === "response.output_item.added" && e.data?.item?.type === "message" + ); + + assert.ok(reasoningDoneIdx >= 0, "reasoning item must be closed"); + assert.ok(messageAddedIdx >= 0, "message item must be added"); + assert.ok( + reasoningDoneIdx < messageAddedIdx, + "reasoning must be closed before message content begins" + ); + + // Reasoning lives at output_index 0; the message must NOT reuse it. + const reasoningIndex = events.find( + (e) => e.event === "response.output_item.added" && e.data?.item?.type === "reasoning" + ).data.output_index; + const messageIndex = events[messageAddedIdx].data.output_index; + + assert.equal(reasoningIndex, 0); + assert.equal(messageIndex, reasoningIndex + 1); + + // The completed snapshot should carry both items at distinct indices. + const completed = events.find((e) => e.event === "response.completed").data.response; + assert.equal(completed.output[0].type, "reasoning"); + assert.equal(completed.output[1].type, "message"); + assert.equal(completed.output[1].content[0].text, "answer"); +}); + +test("translator: native reasoning is closed before message content and uses a distinct output_index", () => { + const state = initState(FORMATS.OPENAI_RESPONSES); + const events = []; + const chunks = [ + { + id: "chatcmpl-1", + model: "gpt-4.1", + choices: [{ index: 0, delta: { reasoning_content: "thinking" }, finish_reason: null }], + }, + { + id: "chatcmpl-1", + model: "gpt-4.1", + choices: [{ index: 0, delta: { content: "answer" }, finish_reason: null }], + }, + { + id: "chatcmpl-1", + model: "gpt-4.1", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + }, + ]; + for (const chunk of chunks) { + const result = openaiToOpenAIResponsesResponse(chunk, state); + if (result) events.push(...result); + } + + const reasoningDoneIdx = events.findIndex( + (e) => e.event === "response.output_item.done" && e.data?.item?.type === "reasoning" + ); + const messageAddedIdx = events.findIndex( + (e) => e.event === "response.output_item.added" && e.data?.item?.type === "message" + ); + + assert.ok(reasoningDoneIdx >= 0, "reasoning item must be closed"); + assert.ok(messageAddedIdx >= 0, "message item must be added"); + assert.ok( + reasoningDoneIdx < messageAddedIdx, + "reasoning must be closed before message content begins" + ); + + const reasoningIndex = events.find( + (e) => e.event === "response.output_item.added" && e.data?.item?.type === "reasoning" + ).data.output_index; + const messageIndex = events[messageAddedIdx].data.output_index; + + assert.equal(reasoningIndex, 0); + assert.equal(messageIndex, reasoningIndex + 1); +});