From 23a8c3c80d697902a8e3afd00480359ffd7b99c3 Mon Sep 17 00:00:00 2001 From: Jordan Ritter Date: Mon, 17 Aug 2026 20:57:42 -0700 Subject: [PATCH] fix: collapse OpenAI Responses-API function-call streams into toolCalls collapseOpenAISSE handled Responses-API reasoning, web-search, and text events but let a catch-all skip every other response.* event, dropping the function-call sequence (output_item.added(function_call), function_call_arguments.delta/.done, output_item.done(function_call)). A tool-call-only turn therefore collapsed to empty content with no toolCalls, recording an empty assistant turn. Accumulate those events into the existing toolCallMap keyed by output_index, mirroring the Chat-Completions tool-call path, and capture the Responses call_id as the tool-call id. Add unit coverage for single/multiple tool-only Responses streams plus a text-only no-regression case. --- src/__tests__/stream-collapse.test.ts | 152 ++++++++++++++++++++++++++ src/stream-collapse.ts | 86 ++++++++++++++- 2 files changed, 237 insertions(+), 1 deletion(-) diff --git a/src/__tests__/stream-collapse.test.ts b/src/__tests__/stream-collapse.test.ts index 0cc50b8e..4b5dee72 100644 --- a/src/__tests__/stream-collapse.test.ts +++ b/src/__tests__/stream-collapse.test.ts @@ -1773,6 +1773,158 @@ describe("collapseOpenAISSE with reasoning", () => { }); }); +describe("collapseOpenAISSE Responses API function calls", () => { + it("collapses a tool-call-only Responses stream into toolCalls (not empty content)", () => { + // The exact event sequence the OpenAI Responses API (and aimock's own + // replay path) emits for a function call: output_item.added(function_call) + // → function_call_arguments.delta* → function_call_arguments.done → + // output_item.done(function_call). Before the fix, all of these were + // dropped by the `response.*` catch-all, yielding empty content and no + // toolCalls. + const body = [ + `data: ${JSON.stringify({ type: "response.created", response: {} })}`, + "", + `data: ${JSON.stringify({ + type: "response.output_item.added", + output_index: 0, + item: { + type: "function_call", + id: "fc_123", + call_id: "call_abc", + name: "get_weather", + arguments: "", + status: "in_progress", + }, + })}`, + "", + `data: ${JSON.stringify({ + type: "response.function_call_arguments.delta", + item_id: "fc_123", + output_index: 0, + delta: '{"ci', + })}`, + "", + `data: ${JSON.stringify({ + type: "response.function_call_arguments.delta", + item_id: "fc_123", + output_index: 0, + delta: 'ty":"Paris"}', + })}`, + "", + `data: ${JSON.stringify({ + type: "response.function_call_arguments.done", + item_id: "fc_123", + output_index: 0, + arguments: '{"city":"Paris"}', + })}`, + "", + `data: ${JSON.stringify({ + type: "response.output_item.done", + output_index: 0, + item: { + type: "function_call", + id: "fc_123", + call_id: "call_abc", + name: "get_weather", + arguments: '{"city":"Paris"}', + status: "completed", + }, + })}`, + "", + `data: ${JSON.stringify({ type: "response.completed", response: {} })}`, + "", + ].join("\n"); + + const result = collapseOpenAISSE(body); + expect(result.toolCalls).toBeDefined(); + expect(result.toolCalls).toHaveLength(1); + expect(result.toolCalls![0].name).toBe("get_weather"); + expect(result.toolCalls![0].arguments).toBe('{"city":"Paris"}'); + expect(JSON.parse(result.toolCalls![0].arguments)).toEqual({ city: "Paris" }); + // The tool-call id is the Responses API `call_id` (what a tool result + // references), not the internal `fc_…` item id. + expect(result.toolCalls![0].id).toBe("call_abc"); + expect(result.content).toBeUndefined(); + }); + + it("collapses multiple Responses function calls keyed by output_index", () => { + const body = [ + `data: ${JSON.stringify({ + type: "response.output_item.added", + output_index: 0, + item: { + type: "function_call", + id: "fc_0", + call_id: "call_0", + name: "lookup", + arguments: "", + }, + })}`, + "", + `data: ${JSON.stringify({ + type: "response.function_call_arguments.delta", + item_id: "fc_0", + output_index: 0, + delta: '{"q":"a"}', + })}`, + "", + `data: ${JSON.stringify({ + type: "response.output_item.added", + output_index: 1, + item: { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "search", + arguments: "", + }, + })}`, + "", + `data: ${JSON.stringify({ + type: "response.function_call_arguments.delta", + item_id: "fc_1", + output_index: 1, + delta: '{"q":"b"}', + })}`, + "", + `data: ${JSON.stringify({ + type: "response.function_call_arguments.done", + item_id: "fc_1", + output_index: 1, + arguments: '{"q":"b"}', + })}`, + "", + ].join("\n"); + + const result = collapseOpenAISSE(body); + expect(result.toolCalls).toHaveLength(2); + expect(result.toolCalls![0]).toMatchObject({ + name: "lookup", + arguments: '{"q":"a"}', + id: "call_0", + }); + expect(result.toolCalls![1]).toMatchObject({ + name: "search", + arguments: '{"q":"b"}', + id: "call_1", + }); + expect(result.content).toBeUndefined(); + }); + + it("does not regress: Responses text-only stream still collapses to content", () => { + const body = [ + `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello" })}`, + "", + `data: ${JSON.stringify({ type: "response.completed", response: {} })}`, + "", + ].join("\n"); + + const result = collapseOpenAISSE(body); + expect(result.content).toBe("Hello"); + expect(result.toolCalls).toBeUndefined(); + }); +}); + describe("collapseAnthropicSSE with thinking", () => { it("extracts reasoning from thinking_delta events", () => { const body = [ diff --git a/src/stream-collapse.ts b/src/stream-collapse.ts index 88918d1e..3228109c 100644 --- a/src/stream-collapse.ts +++ b/src/stream-collapse.ts @@ -442,7 +442,7 @@ export function collapseOpenAISSE(rawBody: string): CollapseResult { continue; } - // Responses API web search events + // Responses API web search + function-call completion events if (parsed.type === "response.output_item.done") { const item = parsed.item as Record | undefined; if (item?.type === "web_search_call") { @@ -452,6 +452,90 @@ export function collapseOpenAISSE(rawBody: string): CollapseResult { continue; } } + if (item?.type === "function_call" && typeof parsed.output_index === "number") { + // Finalize the accumulated tool call. Fill id/name when the opening + // `.added` event was absent, and adopt the full `arguments` ONLY when + // the delta stream produced none — never re-append, since the deltas + // already carry the complete string. + const entry = toolCallMap.get(parsed.output_index); + if (entry) { + if (!entry.id && typeof item.call_id === "string") entry.id = item.call_id; + if (!entry.name && typeof item.name === "string") entry.name = item.name; + if (entry.arguments === "" && typeof item.arguments === "string") { + entry.arguments = item.arguments; + } + } else { + const created = { + id: typeof item.call_id === "string" ? item.call_id : "", + name: typeof item.name === "string" ? item.name : "", + arguments: typeof item.arguments === "string" ? item.arguments : "", + }; + toolCallMap.set(parsed.output_index, created); + orderAtoms.push({ kind: "toolCall", ref: created }); + } + continue; + } + } + + // Responses API function-call streaming events. A tool call arrives as + // `response.output_item.added` (a `function_call` item) → one or more + // `response.function_call_arguments.delta` → `response.function_call_arguments.done` + // (plus the closing `response.output_item.done` handled above). Mirror the + // Chat-Completions tool-call accumulation below so a tool-call-only turn + // collapses to `toolCalls` instead of being silently dropped by the + // `response.*` catch-all. Key by `output_index` (present on every one of + // these events); its integer space is disjoint from the Chat-Completions + // `tool_calls[].index` space because a single stream is never both shapes. + if (parsed.type === "response.output_item.added") { + const item = parsed.item as Record | undefined; + if (item?.type === "function_call" && typeof parsed.output_index === "number") { + if (!toolCallMap.has(parsed.output_index)) { + const created = { + // The Responses API `call_id` is what a tool result references, so + // capture THAT (not the internal `fc_…` item id) as the tool-call + // id — matching what the replay path treats as `toolCall.id`. + id: typeof item.call_id === "string" ? item.call_id : "", + name: typeof item.name === "string" ? item.name : "", + arguments: "", + }; + toolCallMap.set(parsed.output_index, created); + orderAtoms.push({ kind: "toolCall", ref: created }); + } + continue; + } + } + if ( + parsed.type === "response.function_call_arguments.delta" && + typeof parsed.delta === "string" && + typeof parsed.output_index === "number" + ) { + let entry = toolCallMap.get(parsed.output_index); + if (!entry) { + entry = { id: "", name: "", arguments: "" }; + toolCallMap.set(parsed.output_index, entry); + orderAtoms.push({ kind: "toolCall", ref: entry }); + } + entry.arguments += parsed.delta; + continue; + } + if ( + parsed.type === "response.function_call_arguments.done" && + typeof parsed.output_index === "number" + ) { + // The `.done` event repeats the fully-assembled `arguments`; adopt it only + // when the deltas produced nothing (a delta-less stream) so we never + // double-append. + const entry = toolCallMap.get(parsed.output_index); + if (entry) { + if (entry.arguments === "" && typeof parsed.arguments === "string") { + entry.arguments = parsed.arguments; + } + } else if (typeof parsed.arguments === "string") { + const created = { id: "", name: "", arguments: parsed.arguments }; + toolCallMap.set(parsed.output_index, created); + orderAtoms.push({ kind: "toolCall", ref: created }); + } + continue; } // Responses API text content events