fix(sse): close reasoning before message content in Responses stream (#4848)

Integrated into release/v3.8.37 — cherry-picked defining commit onto release tip; CHANGELOG re-merged; tests green.

open-sse/transformer/responsesTransformer.ts

@@ -416,6 +416,14 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv
 
           // Handle text content (may contain <think> 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("<think>")) {
@@ -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;

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("<think>")) {
@@ -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);
     }

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 <think> 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);
+});