diff --git a/open-sse/transformer/responsesTransformer.ts b/open-sse/transformer/responsesTransformer.ts index e98c4d98db..a9c0db949e 100644 --- a/open-sse/transformer/responsesTransformer.ts +++ b/open-sse/transformer/responsesTransformer.ts @@ -209,6 +209,12 @@ export function createResponsesApiTransformStream( funcItemTypes: {}, funcArgsDone: {}, funcItemDone: {}, + // Cached at first computation (see toolCallOutputIndexBase) so every + // added/delta/done event for a given tool call — including ones emitted + // later from the finish_reason handler or flush(), where the reasoning/ + // message state used to derive the base is no longer meaningful to + // recompute — shares exactly the same output_index. + funcOutputIndex: {} as Record, completedOutputItems: [] as Array<{ output_index: number; item: Record; @@ -380,6 +386,27 @@ export function createResponsesApiTransformStream( } }; + // Tool calls sit after reasoning (if any) AND after a text message (if one + // was actually emitted this turn). The provider's own tool_calls[].index is + // scoped only to the tool_calls array and legitimately restarts at 0 — using + // it directly as the Responses API output_index collides with whatever + // reasoning/message item already claimed that slot, and a client that + // tracks response items by output_index silently drops the tool call. + // + // Computed once per tcIdx (from the chunk's own choice index, `chunkIdx`) + // and cached in state.funcOutputIndex so every added/delta/done event for + // that call — including ones emitted later from the finish_reason handler + // or flush(), which have no fresh chunk/reasoning/message state to + // recompute from — shares exactly the same output_index. + const computeToolCallOutputIndex = (chunkIdx, tcIdx) => { + if (state.funcOutputIndex[tcIdx] === undefined) { + const msgIdx = state.reasoningId ? state.reasoningIndex + 1 : chunkIdx; + const base = state.msgItemAdded[msgIdx] ? msgIdx + 1 : msgIdx; + state.funcOutputIndex[tcIdx] = base + normalizeOutputIndex(tcIdx); + } + return state.funcOutputIndex[tcIdx]; + }; + const emitToolCallAdded = (controller, idx) => { if (state.funcItemAdded[idx] || !state.funcCallIds[idx]) return false; @@ -390,7 +417,7 @@ export function createResponsesApiTransformStream( emit(controller, "response.output_item.added", { type: "response.output_item.added", - output_index: idx, + output_index: state.funcOutputIndex[idx], item: { id: `fc_${state.funcCallIds[idx]}`, type: itemType, @@ -406,7 +433,7 @@ export function createResponsesApiTransformStream( const closeToolCall = (controller, idx, recordAsCompleted = true) => { const callId = state.funcCallIds[idx]; if (callId && !state.funcItemDone[idx]) { - const normalizedIndex = normalizeOutputIndex(idx); + const normalizedIndex = state.funcOutputIndex[idx]; let args = state.funcArgsBuf[idx] || "{}"; const toolName = state.funcNames[idx] || ""; emitToolCallAdded(controller, idx); @@ -750,6 +777,7 @@ export function createResponsesApiTransformStream( for (const tc of delta.tool_calls) { const tcIdx = tc.index ?? 0; + const outputIndex = computeToolCallOutputIndex(idx, tcIdx); const newCallId = tc.id; const funcName = tc.function?.name; @@ -765,6 +793,10 @@ export function createResponsesApiTransformStream( delete state.funcItemTypes[tcIdx]; delete state.funcArgsDone[tcIdx]; delete state.funcItemDone[tcIdx]; + // Deliberately keep funcOutputIndex[tcIdx]: the replacement call + // reuses the same positional slot, so it should keep the same + // output_index rather than recomputing (which could drift if + // msgItemAdded state shifted mid-turn). } if (funcName) state.funcNames[tcIdx] = funcName; @@ -786,7 +818,7 @@ export function createResponsesApiTransformStream( emit(controller, "response.function_call_arguments.delta", { type: "response.function_call_arguments.delta", item_id: `fc_${state.funcCallIds[tcIdx]}`, - output_index: tcIdx, + output_index: outputIndex, delta: state.funcArgsBuf[tcIdx], }); } @@ -825,7 +857,7 @@ export function createResponsesApiTransformStream( emit(controller, "response.function_call_arguments.delta", { type: "response.function_call_arguments.delta", item_id: `fc_${refCallId}`, - output_index: tcIdx, + output_index: outputIndex, delta: emittedDelta, }); } diff --git a/tests/unit/responses-transformer-tool-call-reasoning-collision.test.ts b/tests/unit/responses-transformer-tool-call-reasoning-collision.test.ts new file mode 100644 index 0000000000..0568247c4e --- /dev/null +++ b/tests/unit/responses-transformer-tool-call-reasoning-collision.test.ts @@ -0,0 +1,158 @@ +/** + * Regression test for a tool call landing on the same `output_index` as a + * preceding reasoning item in the Responses API stream. + * + * `emitToolCallAdded`/`closeToolCall` in responsesTransformer.ts used the + * provider's raw Chat Completions `tool_calls[].index` directly as the + * Responses API `output_index`. That index is scoped only to the tool_calls + * array and legitimately restarts at 0 for the first tool call — but by the + * time a tool call arrives, a reasoning item (and/or a text message) may + * already have claimed output_index 0 (and 1). A client that tracks response + * items by output_index (as the Responses API spec expects) then sees the + * tool call's added/delta/done events land on an index it already marked + * complete, and silently drops the tool call — producing an "incomplete + * turn" that never dispatches the tool. + * + * Reported live: OpenClaw on combo `default` -> opencode-zen/big-pickle, + * a reasoning block immediately followed by a function call in the same + * turn (no text message in between). + */ + +import test from "node:test"; +import assert from "node:assert/strict"; + +const { createResponsesApiTransformStream } = + await import("../../open-sse/transformer/responsesTransformer.ts"); + +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); + +async function runTransformStream(chunks) { + const stream = createResponsesApiTransformStream(); + 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("tool call immediately after reasoning must not collide on output_index", async () => { + const output = await runTransformStream([ + // Reasoning content — claims output_index 0. + `data: {"id":"chatcmpl-collide","choices":[{"index":0,"delta":{"reasoning_content":"thinking..."}}]}\n\n`, + // A tool call starts. The provider scopes tool_calls[].index to 0 for the + // first (and only) call here, same as reasoning's own output_index. + `data: {"id":"chatcmpl-collide","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"run","arguments":""}}]}}]}\n\n`, + `data: {"id":"chatcmpl-collide","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"cmd\\":\\"ls\\"}"}}]}}]}\n\n`, + `data: {"id":"chatcmpl-collide","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}\n\n`, + ]); + + const events = parseSseOutput(output); + + const reasoningAdded = events.find( + (e) => e.event === "response.output_item.added" && JSON.parse(e.data).item.type === "reasoning" + ); + const toolCallAdded = events.find( + (e) => + e.event === "response.output_item.added" && JSON.parse(e.data).item.type === "function_call" + ); + + assert.ok(reasoningAdded, "reasoning output_item.added must be emitted"); + assert.ok(toolCallAdded, "function_call output_item.added must be emitted"); + + const reasoningIndex = JSON.parse(reasoningAdded.data).output_index; + const toolCallIndex = JSON.parse(toolCallAdded.data).output_index; + + assert.notEqual( + toolCallIndex, + reasoningIndex, + `function_call output_index (${toolCallIndex}) must not collide with reasoning's output_index (${reasoningIndex})` + ); + + // All function_call-related events for this call must share one consistent + // output_index across added/delta/done — a client tracking by output_index + // must be able to follow the whole lifecycle at a single index. + const funcCallEvents = events.filter((e) => { + if ( + e.event !== "response.function_call_arguments.delta" && + e.event !== "response.function_call_arguments.done" && + e.event !== "response.output_item.done" + ) { + return false; + } + if (!e.data) return false; + const parsed = JSON.parse(e.data); + return e.event !== "response.output_item.done" || parsed.item?.type === "function_call"; + }); + assert.ok(funcCallEvents.length > 0, "expected function_call lifecycle events"); + for (const e of funcCallEvents) { + assert.equal( + JSON.parse(e.data).output_index, + toolCallIndex, + `event ${e.event} must use the same output_index as the tool call's added event` + ); + } + + // response.completed output must contain both items at distinct indices. + const completed = JSON.parse(events.find((e) => e.event === "response.completed").data).response; + const reasoningItem = completed.output.find((item) => item.type === "reasoning"); + const funcItem = completed.output.find((item) => item.type === "function_call"); + assert.ok(reasoningItem, "completed output must include the reasoning item"); + assert.ok(funcItem, "completed output must include the function_call item"); + assert.equal(funcItem.call_id, "call_1"); + assert.equal(funcItem.arguments, '{"cmd":"ls"}'); +}); + +test("multiple tool calls after reasoning use sequential output_index values, none colliding with reasoning", async () => { + const output = await runTransformStream([ + `data: {"id":"chatcmpl-multi","choices":[{"index":0,"delta":{"reasoning_content":"planning"}}]}\n\n`, + `data: {"id":"chatcmpl-multi","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_a","function":{"name":"first","arguments":"{}"}}]}}]}\n\n`, + `data: {"id":"chatcmpl-multi","choices":[{"index":0,"delta":{"tool_calls":[{"index":1,"id":"call_b","function":{"name":"second","arguments":"{}"}}]}}]}\n\n`, + `data: {"id":"chatcmpl-multi","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}\n\n`, + ]); + + const events = parseSseOutput(output); + const addedEvents = events.filter((e) => e.event === "response.output_item.added"); + const indexByType = addedEvents.map((e) => { + const parsed = JSON.parse(e.data); + return { type: parsed.item.type, output_index: parsed.output_index }; + }); + + const reasoningIdx = indexByType.find((i) => i.type === "reasoning").output_index; + const funcIndices = indexByType.filter((i) => i.type === "function_call").map((i) => i.output_index); + + assert.equal(funcIndices.length, 2); + assert.ok(new Set(funcIndices).size === 2, "the two tool calls must not share an output_index"); + for (const fi of funcIndices) { + assert.notEqual(fi, reasoningIdx, "no tool call may collide with the reasoning output_index"); + } +});