diff --git a/changelog.d/fixes/11144-responses-parallel-tool-calls-index.md b/changelog.d/fixes/11144-responses-parallel-tool-calls-index.md new file mode 100644 index 0000000000..35ba19b279 --- /dev/null +++ b/changelog.d/fixes/11144-responses-parallel-tool-calls-index.md @@ -0,0 +1 @@ +- **fix(sse):** parallel `function_call` items in a Responses API stream (e.g. several tool calls dispatched in the same turn) now each get a stable, distinct `index`/`id` when translated to Chat Completions streaming deltas, instead of colliding on index 0 and tripping strict stream parsers with `Expected 'id' to be a string.` ([#11144](https://github.com/diegosouzapw/OmniRoute/pull/11144)) diff --git a/open-sse/translator/response/openai-responses.ts b/open-sse/translator/response/openai-responses.ts index 0c59fac4fb..01ad55d72f 100644 --- a/open-sse/translator/response/openai-responses.ts +++ b/open-sse/translator/response/openai-responses.ts @@ -866,21 +866,25 @@ export function openaiResponsesToOpenAIResponse(chunk, state) { function openaiResponsesToOpenAIResponseStream(chunk, state) { if (!chunk) { - if ( - state.currentToolCallNeedsNormalization && - state.currentToolCallArgsBuffer && - state.currentToolCallName - ) { - const toolSchema = state.toolSchemas?.get(state.currentToolCallName); - const argsToEmit = stripEmptyOptionalToolArgs( - state.currentToolCallArgsBuffer, - state.currentToolCallName, - toolSchema - ); - const argsStr = - typeof argsToEmit === "string" ? argsToEmit : JSON.stringify(argsToEmit ?? {}); - state.currentToolCallArgsBuffer = ""; - state.currentToolCallNeedsNormalization = false; + // Iterate every still-open call needing schema-aware normalization, not just a + // single one — multiple parallel calls can each be pending here if the stream + // ends before their output_item.done arrives. + const pendingNormalized: Array<{ index: number; argsStr: string }> = []; + if (state.toolCallByCallId instanceof Map) { + for (const entry of state.toolCallByCallId.values()) { + if (entry.needsNormalization && entry.argsBuffer) { + const toolSchema = state.toolSchemas?.get(entry.name); + const argsToEmit = stripEmptyOptionalToolArgs(entry.argsBuffer, entry.name, toolSchema); + pendingNormalized.push({ + index: entry.index, + argsStr: typeof argsToEmit === "string" ? argsToEmit : JSON.stringify(argsToEmit ?? {}), + }); + entry.argsBuffer = ""; + entry.needsNormalization = false; + } + } + } + if (pendingNormalized.length > 0) { state.finishReasonSent = true; state.finishReason = "tool_calls"; const common = { @@ -889,24 +893,21 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { created: state.created, model: state.model || "gpt-4", }; - return [ - { - ...common, - choices: [ - { - index: 0, - delta: { - tool_calls: [{ index: state.toolCallIndex, function: { arguments: argsStr } }], - }, - finish_reason: null, - }, - ], - }, - { - ...common, - choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }], - }, - ]; + const chunks: Record[] = pendingNormalized.map(({ index, argsStr }) => ({ + ...common, + choices: [ + { + index: 0, + delta: { tool_calls: [{ index, function: { arguments: argsStr } }] }, + finish_reason: null, + }, + ], + })); + chunks.push({ + ...common, + choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }], + }); + return chunks; } // Flush: send final chunk with finish_reason if (!state.finishReasonSent && state.started) { @@ -952,7 +953,23 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { state.chatId = `chatcmpl-${Date.now()}`; state.created = Math.floor(Date.now() / 1000); state.toolCallIndex = 0; + // Kept for computeFinishReason (synthesizeCompletedToolCalls.ts) compatibility — + // that snapshot path mutates it directly and expects it to exist. In a turn with + // multiple parallel calls this only ever reflects the LAST one opened/closed, so + // it must never be used to identify a specific call — only as the "is at least + // one tool call in flight this turn" signal computeFinishReason needs, which + // toolCallIndex > 0 already covers on its own once any call has been added. state.currentToolCallId = null; + // Per-call state keyed by call_id (replaces the old singular + // currentToolCallId/ArgsBuffer/Name/NeedsNormalization/Deferred fields, which + // assumed only one function_call could ever be in flight at a time). + state.toolCallByCallId = new Map(); + // response.function_call_arguments.delta carries `item_id`/`output_index`, not + // `call_id` — resolve either one back to the call_id key used by + // toolCallByCallId (two independent reverse maps, since some upstreams omit + // item_id on delta events but still send output_index). + state.toolCallItemToCallId = new Map(); + state.toolCallOutputIndexToCallId = new Map(); } // Text content delta @@ -983,22 +1000,48 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { // Function call started if (eventType === "response.output_item.added" && data.item?.type === "function_call") { const item = data.item; - state.currentToolCallId = item.call_id || fallbackToolCallId(); - state.currentToolCallArgsBuffer = ""; // reset per-call arg buffer - state.currentToolCallDeferred = false; + const callId = item.call_id || fallbackToolCallId(); + // Kept for computeFinishReason (synthesizeCompletedToolCalls.ts) compatibility. + state.currentToolCallId = callId; + + const toolName = normalizeToolName(item.name); + // Assign this call's index NOW, at .added, not at .done — two calls opened before + // either closes (a genuine parallel dispatch) must never share an index. Deferred + // (still-nameless) calls are the one exception: they don't claim an index until + // .done resolves a real name, so a call that never gets one never burns a slot + // another call could have used. + let index: number | null = null; + if (toolName) { + index = state.toolCallIndex ?? 0; + state.toolCallIndex = index + 1; + } + + if (!(state.toolCallByCallId instanceof Map)) state.toolCallByCallId = new Map(); + state.toolCallByCallId.set(callId, { + index, + name: toolName, + argsBuffer: "", + deferred: !toolName, + needsNormalization: toolName === "Agent", + }); + if (!(state.toolCallItemToCallId instanceof Map)) state.toolCallItemToCallId = new Map(); + if (item.id) state.toolCallItemToCallId.set(item.id, callId); + // `output_index` is a top-level field on every Responses API streamed event + // (response.output_item.added/.done AND function_call_arguments.delta alike) — + // an identifier independent of item_id, for upstreams that omit item_id on delta + // events. + if (!(state.toolCallOutputIndexToCallId instanceof Map)) { + state.toolCallOutputIndexToCallId = new Map(); + } + if (data.output_index != null) state.toolCallOutputIndexToCallId.set(data.output_index, callId); // Track this call_id so response.completed doesn't synthesize a duplicate if (!state.toolCallIdsSeen) state.toolCallIdsSeen = new Set(); - if (state.currentToolCallId) state.toolCallIdsSeen.add(state.currentToolCallId); + state.toolCallIdsSeen.add(callId); - const toolName = normalizeToolName(item.name); - state.currentToolName = toolName; // track for schema lookup at done time - state.currentToolCallName = toolName; - state.currentToolCallNeedsNormalization = toolName === "Agent"; if (!toolName) { // Some Responses providers briefly emit placeholder/empty tool names. // Defer emission until output_item.done in case the final name is populated there. - state.currentToolCallDeferred = true; return null; } @@ -1013,8 +1056,8 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { delta: { tool_calls: [ { - index: state.toolCallIndex, - id: state.currentToolCallId, + index, + id: callId, type: "function", function: { name: toolName, @@ -1037,11 +1080,26 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { const argsDelta = data.delta || ""; if (!argsDelta) return null; - state.currentToolCallArgsBuffer = (state.currentToolCallArgsBuffer || "") + argsDelta; - if (state.currentToolCallDeferred || state.currentToolCallNeedsNormalization) return null; + // Resolve which in-flight call this delta belongs to. Try item_id first (the + // field the Responses API documents for this event), then output_index (also a + // top-level field on this event, and independent of item_id — covers upstreams + // that omit item_id on delta events but still send output_index). Only once both + // identifying fields are absent/unresolved do we fall back to guessing (the + // single open call, or the most recently opened one as a last resort). + const map = state.toolCallByCallId instanceof Map ? state.toolCallByCallId : null; + let callId = data.item_id ? state.toolCallItemToCallId?.get(data.item_id) : undefined; + if (!callId && data.output_index != null) { + callId = state.toolCallOutputIndexToCallId?.get(data.output_index); + } + if (!callId && map) { + callId = map.size === 1 ? [...map.keys()][0] : state.currentToolCallId; + } + const entry = callId ? map?.get(callId) : undefined; + if (!entry) return null; // #9168: buffer arguments until output_item.done for schema-aware null normalization // Previously emitted raw null values for optional enum fields (e.g. isolation: null). + entry.argsBuffer = (entry.argsBuffer || "") + argsDelta; return null; } @@ -1061,13 +1119,30 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { // carry the complete arguments only in output_item.done (no preceding delta events). if (eventType === "response.output_item.done" && data.item?.type === "function_call") { const item = data.item; - const buffered = state.currentToolCallArgsBuffer || ""; - const currentIndex = state.toolCallIndex; // capture before increment - const callId = item.call_id || state.currentToolCallId || fallbackToolCallId(); + const map = state.toolCallByCallId instanceof Map ? state.toolCallByCallId : null; + let callId = item.call_id; + if (!callId && item.id) callId = state.toolCallItemToCallId?.get(item.id); + if (!callId) callId = state.currentToolCallId || fallbackToolCallId(); + const trackedEntry = callId ? map?.get(callId) : undefined; + // Some upstreams (e.g. Codex) send the complete payload only in output_item.done, + // with no preceding output_item.added at all — there is no tracked entry to read an + // index from. + const entry = trackedEntry || { index: null, argsBuffer: "", deferred: false }; + + const buffered = entry.argsBuffer || ""; const toolName = normalizeToolName(item.name); + + // Claim (and advance) this call's index now if it wasn't assigned at .added — either + // a deferred call whose name has just now resolved, or a Codex-style done-only + // payload that never had an .added at all. A deferred call whose name is STILL empty + // never claims an index (nothing was ever emitted for it either way). + if (entry.index == null && toolName) { + entry.index = state.toolCallIndex ?? 0; + state.toolCallIndex = entry.index + 1; + } + const currentIndex = entry.index; const toolSchema = state.toolSchemas?.get(toolName); const shouldNormalizeArguments = toolName === "Agent"; - state.currentToolCallNeedsNormalization = shouldNormalizeArguments; if (toolName && state.toolCalls instanceof Map) { const completedArguments = @@ -1077,6 +1152,9 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { toolName, toolSchema ); + // Keyed by index, not insertion order — readers that need call order for + // parallel calls closed out of order should sort by this key rather than + // relying on Map iteration order. state.toolCalls.set(currentIndex, { id: callId, index: currentIndex, @@ -1095,17 +1173,17 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { if (!state.toolCallIdsSeen) state.toolCallIdsSeen = new Set(); if (callId) state.toolCallIdsSeen.add(callId); - if (state.currentToolCallDeferred) { - state.currentToolCallDeferred = false; - state.currentToolCallArgsBuffer = ""; - state.currentToolCallId = null; + // This call is fully closed — remove it from the in-flight map (bounds the map + // to genuinely in-flight calls, and keeps the single-open-call fallback in the + // function_call_arguments.delta handler correct for whichever call opens next). + if (map && callId) map.delete(callId); + if (state.currentToolCallId === callId) state.currentToolCallId = null; + if (entry.deferred) { if (!toolName) { return null; } - state.toolCallIndex++; - const terminalArguments = typeof item.arguments === "string" ? item.arguments.length > 0 @@ -1148,12 +1226,7 @@ function openaiResponsesToOpenAIResponseStream(chunk, state) { }; } - state.toolCallIndex++; - state.currentToolCallArgsBuffer = ""; // reset for next tool call - state.currentToolCallId = null; - const needsNormalization = state.currentToolCallNeedsNormalization === true; - state.currentToolCallNeedsNormalization = false; - state.currentToolCallName = ""; + const needsNormalization = shouldNormalizeArguments; // Nullable omission sentinels must be normalized before any argument bytes reach the client. // Other tool calls retain immediate argument streaming. diff --git a/tests/unit/responses-parallel-tool-calls-index.test.ts b/tests/unit/responses-parallel-tool-calls-index.test.ts new file mode 100644 index 0000000000..f825f8700b --- /dev/null +++ b/tests/unit/responses-parallel-tool-calls-index.test.ts @@ -0,0 +1,294 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { openaiResponsesToOpenAIResponse } = + await import("../../open-sse/translator/response/openai-responses.ts"); + +// Issue: 2+ `function_call` items opened (response.output_item.added) before any of +// them closes (response.output_item.done) — a genuine parallel tool-call dispatch — +// causes `state.toolCallIndex` (only incremented in the `.done` handler) to stay at 0 +// for every "added" header chunk. Clients that key their tool-call accumulator by +// `delta.tool_calls[].index` (e.g. opencode's github-copilot chat-language-model +// stream parser) then see the *first* `.done` argument chunk at index 1/2 with no +// prior header and no `id`, and throw "Expected 'id' to be a string." +test("Responses -> OpenAI: parallel function_call items get distinct index+id on the added header", () => { + const state = {}; + + const added0 = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_0", name: "task" }, + }, + state + ); + const added1 = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_1", name: "task" }, + }, + state + ); + const added2 = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_2", name: "task" }, + }, + state + ); + + const headers = [added0, added1, added2].map((r) => r.choices[0].delta.tool_calls[0]); + + assert.deepEqual( + headers.map((h) => h.index), + [0, 1, 2], + "each parallel tool call must get its own header index, not all 0" + ); + assert.deepEqual( + headers.map((h) => h.id), + ["call_0", "call_1", "call_2"] + ); + + const done0 = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_0", name: "task", arguments: '{"i":0}' }, + }, + state + ); + const done1 = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_1", name: "task", arguments: '{"i":1}' }, + }, + state + ); + const done2 = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_2", name: "task", arguments: '{"i":2}' }, + }, + state + ); + + assert.deepEqual( + [done0, done1, done2].map((r) => r.choices[0].delta.tool_calls[0].index), + [0, 1, 2], + "argument chunks must reuse the SAME index assigned at .added time for each call_id" + ); +}); + +test("Responses -> OpenAI: parallel calls closed out of order keep their own index", () => { + const state = {}; + + for (const callId of ["call_a", "call_b", "call_c"]) { + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: callId, name: "task" }, + }, + state + ); + } + + // Close in reverse order: c, then a, then b. + const doneC = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_c", name: "task", arguments: "{}" }, + }, + state + ); + const doneA = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_a", name: "task", arguments: "{}" }, + }, + state + ); + const doneB = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_b", name: "task", arguments: "{}" }, + }, + state + ); + + assert.equal(doneC.choices[0].delta.tool_calls[0].index, 2); + assert.equal(doneA.choices[0].delta.tool_calls[0].index, 0); + assert.equal(doneB.choices[0].delta.tool_calls[0].index, 1); +}); + +test("Responses -> OpenAI: argument deltas interleaved across 2 parallel calls do not get glued together", () => { + const state = {}; + + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_x", name: "Read", id: "fc_call_x" }, + }, + state + ); + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_y", name: "Read", id: "fc_call_y" }, + }, + state + ); + + // Interleave argument deltas by item_id — x, y, x, y — before either closes. + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", item_id: "fc_call_x", delta: '{"filePath"' }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", item_id: "fc_call_y", delta: '{"filePath"' }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", item_id: "fc_call_x", delta: ':"/a.txt"}' }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", item_id: "fc_call_y", delta: ':"/b.txt"}' }, + state + ); + + const doneX = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_x", name: "Read" }, + }, + state + ); + const doneY = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_y", name: "Read" }, + }, + state + ); + + assert.equal(doneX.choices[0].delta.tool_calls[0].function.arguments, '{"filePath":"/a.txt"}'); + assert.equal(doneY.choices[0].delta.tool_calls[0].function.arguments, '{"filePath":"/b.txt"}'); +}); + +test("Responses -> OpenAI: a deferred (nameless) call that never resolves a name never consumes an index", () => { + const state = {}; + + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_deferred", name: "" }, + }, + state + ); + const done = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_deferred", name: " " }, + }, + state + ); + + assert.equal(done, null); + assert.equal(state.toolCallIndex, 0); +}); + +test("Responses -> OpenAI: argument deltas interleaved across 2 parallel calls resolve by output_index when the upstream omits item_id", () => { + const state = {}; + + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + output_index: 0, + item: { type: "function_call", call_id: "call_p", name: "Read" }, + }, + state + ); + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + output_index: 1, + item: { type: "function_call", call_id: "call_q", name: "Read" }, + }, + state + ); + + // No item_id on any of these deltas — only output_index, which the Responses API + // guarantees on every streamed event regardless of whether item_id is also sent. + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", output_index: 0, delta: '{"filePath"' }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", output_index: 1, delta: '{"filePath"' }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", output_index: 0, delta: ':"/p.txt"}' }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", output_index: 1, delta: ':"/q.txt"}' }, + state + ); + + const doneP = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_p", name: "Read" }, + }, + state + ); + const doneQ = openaiResponsesToOpenAIResponse( + { + type: "response.output_item.done", + item: { type: "function_call", call_id: "call_q", name: "Read" }, + }, + state + ); + + assert.equal(doneP.choices[0].delta.tool_calls[0].function.arguments, '{"filePath":"/p.txt"}'); + assert.equal(doneQ.choices[0].delta.tool_calls[0].function.arguments, '{"filePath":"/q.txt"}'); +}); + +test("Responses -> OpenAI: 2 parallel Agent calls still open at stream end each get their own flush chunk", () => { + const state = {}; + + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_agent0", name: "Agent" }, + }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", output_index: 0, delta: '{"task":"a"}' }, + state + ); + openaiResponsesToOpenAIResponse( + { + type: "response.output_item.added", + item: { type: "function_call", call_id: "call_agent1", name: "Agent" }, + }, + state + ); + openaiResponsesToOpenAIResponse( + { type: "response.function_call_arguments.delta", output_index: 1, delta: '{"task":"b"}' }, + state + ); + + // Stream ends (chunk === null) before either call's output_item.done arrives. + const flushed = openaiResponsesToOpenAIResponse(null, state); + + assert.ok(Array.isArray(flushed)); + const argChunks = flushed.filter((c) => c.choices[0].delta.tool_calls); + assert.deepEqual( + argChunks.map((c) => c.choices[0].delta.tool_calls[0].index).sort(), + [0, 1], + "each still-open parallel call must get its own flush chunk, at its own index" + ); + const finalChunk = flushed[flushed.length - 1]; + assert.equal(finalChunk.choices[0].finish_reason, "tool_calls"); +});