diff --git a/src/chat/outbound.ts b/src/chat/outbound.ts index ed3928b0ede..cebeaacdb8e 100644 --- a/src/chat/outbound.ts +++ b/src/chat/outbound.ts @@ -132,7 +132,10 @@ export function responsesSseToChatCompletionsSse( // tool call_id -> streaming index (OpenAI requires stable indices per tool call) const toolIndexByCallId = new Map(); const toolIndexByItemId = new Map(); + const toolCallIdByIndex = new Map(); const toolNameByIndex = new Map(); + const toolArgumentsByIndex = new Map(); + const emittedToolIndexes = new Set(); let nextToolIndex = 0; let sseIterator: AsyncGenerator<{ event?: string; data: string }> | undefined; const upstreamAbort = new AbortController(); @@ -166,8 +169,47 @@ export function responsesSseToChatCompletionsSse( frame.choices = [{ index: 0, delta: { reasoning_content: text }, finish_reason: null }]; emit(frame); }; + const resolveFinalArguments = (candidate: unknown, streamed: string) => { + if (typeof candidate !== "string") return streamed; + // Some compatible providers send an empty final snapshot. Do not let that erase + // non-empty streamed arguments; genuinely empty calls have no buffered content. + return candidate.length > 0 || streamed.length === 0 ? candidate : streamed; + }; + const emitToolCall = (toolIndex: number, callId: string, name: string, args: string) => { + if (!callId || emittedToolIndexes.has(toolIndex)) return; + emittedToolIndexes.add(toolIndex); + ensureRole(); + const frame = chunkBase(id, model, created); + frame.choices = [{ + index: 0, + delta: { + tool_calls: [{ + index: toolIndex, + id: callId, + type: "function", + function: { name, arguments: args }, + }], + }, + finish_reason: null, + }]; + emit(frame); + }; + const flushPendingToolCalls = () => { + const pending = [...toolCallIdByIndex.entries()].sort(([a], [b]) => a - b); + for (const [toolIndex, callId] of pending) { + emitToolCall( + toolIndex, + callId, + toolNameByIndex.get(toolIndex) ?? "", + toolArgumentsByIndex.get(toolIndex) ?? "", + ); + } + }; const finish = (finishReason: string, usage: unknown) => { if (terminated) return; + // A valid completed/incomplete terminal frame may arrive without output_item.done. + // Preserve any known tool call before emitting its finish reason. + flushPendingToolCalls(); terminated = true; ensureRole(); const frame = chunkBase(id, model, created); @@ -232,24 +274,15 @@ export function responsesSseToChatCompletionsSse( toolIndex = nextToolIndex++; toolIndexByCallId.set(callId, toolIndex); } + toolCallIdByIndex.set(toolIndex, callId); if (typeof item.id === "string") toolIndexByItemId.set(item.id, toolIndex); if (name) toolNameByIndex.set(toolIndex, name); - const frame = chunkBase(id, model, created); - frame.choices = [{ - index: 0, - delta: { - tool_calls: [{ - index: toolIndex, - id: callId, - type: "function", - // Always include name so clients that replace (not merge) tool_calls - // deltas never end up with function.name-less history. - function: { name, arguments: "" }, - }], - }, - finish_reason: null, - }]; - emit(frame); + if (!toolArgumentsByIndex.has(toolIndex)) { + toolArgumentsByIndex.set( + toolIndex, + typeof item.arguments === "string" ? item.arguments : "", + ); + } break; } case "response.function_call_arguments.delta": { @@ -258,23 +291,24 @@ export function responsesSseToChatCompletionsSse( const toolIndex = (itemId ? toolIndexByItemId.get(itemId) : undefined) ?? (nextToolIndex > 0 ? nextToolIndex - 1 : 0); ensureRole(); - const knownName = toolNameByIndex.get(toolIndex) ?? ""; - const fn: Rec = { arguments: data.delta }; - // Re-emit name on every args delta: GitHub Copilot App / some ChatGPT clients - // replace the whole function object instead of merging, which otherwise drops name. - if (knownName) fn.name = knownName; - const frame = chunkBase(id, model, created); - frame.choices = [{ - index: 0, - delta: { - tool_calls: [{ - index: toolIndex, - function: fn, - }], - }, - finish_reason: null, - }]; - emit(frame); + toolArgumentsByIndex.set( + toolIndex, + (toolArgumentsByIndex.get(toolIndex) ?? "") + data.delta, + ); + break; + } + case "response.function_call_arguments.done": { + const itemId = typeof data.item_id === "string" ? data.item_id : undefined; + const toolIndex = itemId ? toolIndexByItemId.get(itemId) : undefined; + if (toolIndex === undefined) break; + sawToolUse = true; + const name = typeof data.name === "string" ? data.name : ""; + if (name) toolNameByIndex.set(toolIndex, name); + const streamedArgs = toolArgumentsByIndex.get(toolIndex) ?? ""; + toolArgumentsByIndex.set( + toolIndex, + resolveFinalArguments(data.arguments, streamedArgs), + ); break; } case "response.output_item.done": { @@ -284,34 +318,25 @@ export function responsesSseToChatCompletionsSse( sawToolUse = true; const callId = typeof item.call_id === "string" ? item.call_id : ""; const name = typeof item.name === "string" ? item.name : ""; - const args = typeof item.arguments === "string" ? item.arguments : ""; if (!callId) break; - const existingIndex = toolIndexByCallId.get(callId); - const isNew = existingIndex === undefined; + const itemId = typeof item.id === "string" ? item.id : undefined; + const existingIndex = toolIndexByCallId.get(callId) + ?? (itemId ? toolIndexByItemId.get(itemId) : undefined); const toolIndex = existingIndex ?? nextToolIndex++; - if (isNew) toolIndexByCallId.set(callId, toolIndex); - if (typeof item.id === "string") toolIndexByItemId.set(item.id, toolIndex); + toolIndexByCallId.set(callId, toolIndex); + toolCallIdByIndex.set(toolIndex, callId); + if (itemId) toolIndexByItemId.set(itemId, toolIndex); if (name) toolNameByIndex.set(toolIndex, name); const finalName = name || toolNameByIndex.get(toolIndex) || ""; - // Last-write-wins: the done-frame snapshot is authoritative final arguments. - // Always re-emit full identity (id/type/name) so replace-style clients keep function.name. - if (args.length > 0 || isNew || finalName) { - ensureRole(); - const frame = chunkBase(id, model, created); - frame.choices = [{ - index: 0, - delta: { - tool_calls: [{ - index: toolIndex, - id: callId, - type: "function", - function: { name: finalName, arguments: args }, - }], - }, - finish_reason: null, - }]; - emit(frame); - } + const streamedArgs = toolArgumentsByIndex.get(toolIndex) ?? ""; + // A Responses stream can carry incremental/finalized argument events plus an + // authoritative output_item.done snapshot. Chat Completions tool-call fields are + // append-only deltas, so forwarding multiple representations corrupts clients that + // accumulate them. Emit one complete call here; finish() flushes any item whose + // done event was omitted, and replace-style clients still get a complete object. + const args = resolveFinalArguments(item.arguments, streamedArgs); + toolArgumentsByIndex.set(toolIndex, args); + emitToolCall(toolIndex, callId, finalName, args); } break; } diff --git a/tests/chat-completions-endpoint.test.ts b/tests/chat-completions-endpoint.test.ts index 2599defa5b9..17c600202c4 100644 --- a/tests/chat-completions-endpoint.test.ts +++ b/tests/chat-completions-endpoint.test.ts @@ -66,6 +66,40 @@ function mockConfig(baseUrl: string): OcxConfig { } as OcxConfig; } +type StreamedToolCall = { + index?: number; + id?: string; + type?: string; + function?: { name?: string; arguments?: string }; +}; + +type ChatStreamChunk = { + choices?: Array<{ + delta?: { tool_calls?: StreamedToolCall[] }; + finish_reason?: string | null; + }>; +}; + +async function convertResponsesFrames(frames: string[], model = "gpt-test") { + const { responsesSseToChatCompletionsSse } = await import("../src/chat/outbound"); + const stream = responsesSseToChatCompletionsSse(new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + for (const frame of frames) controller.enqueue(encoder.encode(frame)); + controller.close(); + }, + }), model); + const text = await new Response(stream).text(); + const chunks = text.split("\n\n") + .map(block => block.trim()) + .filter(block => block.startsWith("data: ") && !block.includes("[DONE]")) + .map(block => JSON.parse(block.slice(6)) as ChatStreamChunk); + return { + chunks, + toolCalls: chunks.flatMap(chunk => chunk.choices?.[0]?.delta?.tool_calls ?? []), + }; +} + test("chatCompletionsToResponsesBody maps messages/tools/system", () => { const body = chatCompletionsToResponsesBody({ model: "mock/test-model", @@ -224,36 +258,31 @@ test("chatCompletionsToResponsesBody maps response_format and rejects unknown ty })).toThrow(ChatCompletionsRequestError); }); -test("responsesSseToChatCompletionsSse routes parallel tool argument deltas by item_id", async () => { - const { responsesSseToChatCompletionsSse } = await import("../src/chat/outbound"); +test("responsesSseToChatCompletionsSse emits parallel tool calls once with stable indices", async () => { const frames = [ `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", output_index: 0, item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "alpha", arguments: "" } })}\n\n`, `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", output_index: 1, item: { type: "function_call", id: "fc_b", call_id: "call_b", name: "beta", arguments: "" } })}\n\n`, `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '{"a":' })}\n\n`, `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_b", delta: '{"b":' })}\n\n`, `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: "1}" })}\n\n`, + `event: response.output_item.done\ndata: ${JSON.stringify({ type: "response.output_item.done", output_index: 0, item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "alpha", arguments: '{"a":1}' } })}\n\n`, + `event: response.output_item.done\ndata: ${JSON.stringify({ type: "response.output_item.done", output_index: 1, item: { type: "function_call", id: "fc_b", call_id: "call_b", name: "beta", arguments: '{"b":2}' } })}\n\n`, `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } })}\n\n`, ]; - const stream = responsesSseToChatCompletionsSse(new ReadableStream({ - start(controller) { - const enc = new TextEncoder(); - for (const frame of frames) controller.enqueue(enc.encode(frame)); - controller.close(); + const { toolCalls } = await convertResponsesFrames(frames, "mock/test-model"); + expect(toolCalls).toEqual([ + { + index: 0, + id: "call_a", + type: "function", + function: { name: "alpha", arguments: '{"a":1}' }, + }, + { + index: 1, + id: "call_b", + type: "function", + function: { name: "beta", arguments: '{"b":2}' }, }, - }), "mock/test-model"); - const text = await new Response(stream).text(); - const chunks = text.split("\n\n") - .map(block => block.trim()) - .filter(block => block.startsWith("data: ") && !block.includes("[DONE]")) - .map(block => JSON.parse(block.slice(6)) as { - choices?: Array<{ delta?: { tool_calls?: Array<{ index?: number; function?: { arguments?: string } }> } }>; - }); - const argDeltas = chunks.flatMap(chunk => chunk.choices?.[0]?.delta?.tool_calls ?? []) - .filter(tc => typeof tc.function?.arguments === "string" && tc.function.arguments.length > 0); - expect(argDeltas.map(tc => ({ index: tc.index, arguments: tc.function?.arguments }))).toEqual([ - { index: 0, arguments: '{"a":' }, - { index: 1, arguments: '{"b":' }, - { index: 0, arguments: "1}" }, ]); }); @@ -611,7 +640,7 @@ test("streaming /v1/chat/completions does not clean-DONE after response.failed", }); -test("responsesSseToChatCompletionsSse always re-emits function.name on args/done deltas", async () => { +test("responsesSseToChatCompletionsSse emits one complete named tool call", async () => { const { responsesSseToChatCompletionsSse, collectChatCompletion } = await import("../src/chat/outbound"); const frames = [ `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: "" } })}\n\n`, @@ -620,29 +649,16 @@ test("responsesSseToChatCompletionsSse always re-emits function.name on args/don `event: response.output_item.done\ndata: ${JSON.stringify({ type: "response.output_item.done", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: '{"cmd":"ls"}' } })}\n\n`, `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { status: "completed", usage: { input_tokens: 1, output_tokens: 1 } } })}\n\n`, ]; - const stream = responsesSseToChatCompletionsSse(new ReadableStream({ - start(controller) { - const enc = new TextEncoder(); - for (const frame of frames) controller.enqueue(enc.encode(frame)); - controller.close(); - }, - }), "gpt-test"); - const text = await new Response(stream).text(); - const chunks = text.split("\n\n") - .map(block => block.trim()) - .filter(block => block.startsWith("data: ") && !block.includes("[DONE]")) - .map(block => JSON.parse(block.slice(6)) as { - choices?: Array<{ delta?: { tool_calls?: Array<{ function?: { name?: string; arguments?: string } }> } }>; - }); - const toolDeltas = chunks.flatMap(c => c.choices?.[0]?.delta?.tool_calls ?? []) - .filter(tc => tc.function && ("arguments" in (tc.function ?? {}) || "name" in (tc.function ?? {}))); - // Every tool delta that carries arguments must also carry the name for replace-style clients. - for (const tc of toolDeltas) { - if (typeof tc.function?.arguments === "string") { - expect(tc.function?.name).toBe("exec_command"); - } - } - // Last-write-wins collect still yields a complete named tool call. + const { toolCalls: toolDeltas } = await convertResponsesFrames(frames); + // Responses emits two argument deltas plus a full done snapshot. Chat Completions clients + // append function fields, so expose one complete tool-call delta rather than all three. + expect(toolDeltas).toEqual([{ + index: 0, + id: "call_a", + type: "function", + function: { name: "exec_command", arguments: '{"cmd":"ls"}' }, + }]); + // Buffered non-stream collection still yields the same complete named tool call. const completion = await collectChatCompletion(responsesSseToChatCompletionsSse(new ReadableStream({ start(controller) { const enc = new TextEncoder(); @@ -656,6 +672,98 @@ test("responsesSseToChatCompletionsSse always re-emits function.name on args/don expect(toolCalls[0]?.function?.arguments).toBe('{"cmd":"ls"}'); }); +test("responsesSseToChatCompletionsSse falls back to buffered arguments when the done item omits them", async () => { + const frames = [ + `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: "" } })}\n\n`, + `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '{"cmd":' })}\n\n`, + `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '"pwd"}' })}\n\n`, + `event: response.output_item.done\ndata: ${JSON.stringify({ type: "response.output_item.done", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command" } })}\n\n`, + `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { status: "completed" } })}\n\n`, + ]; + const { toolCalls } = await convertResponsesFrames(frames); + expect(toolCalls).toEqual([{ + index: 0, + id: "call_a", + type: "function", + function: { name: "exec_command", arguments: '{"cmd":"pwd"}' }, + }]); +}); + +test("responsesSseToChatCompletionsSse flushes buffered tool calls before an incomplete terminal frame", async () => { + const frames = [ + `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: "" } })}\n\n`, + `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '{"cmd":' })}\n\n`, + `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '"pwd"}' })}\n\n`, + `event: response.incomplete\ndata: ${JSON.stringify({ type: "response.incomplete", response: { status: "incomplete", incomplete_details: { reason: "max_output_tokens" } } })}\n\n`, + ]; + const { chunks, toolCalls } = await convertResponsesFrames(frames); + expect(toolCalls).toEqual([{ + index: 0, + id: "call_a", + type: "function", + function: { name: "exec_command", arguments: '{"cmd":"pwd"}' }, + }]); + const toolChunkIndex = chunks.findIndex(chunk => + (chunk.choices?.[0]?.delta?.tool_calls?.length ?? 0) > 0 + ); + const finishChunkIndex = chunks.findIndex(chunk => + chunk.choices?.[0]?.finish_reason === "length" + ); + expect(toolChunkIndex).toBeGreaterThanOrEqual(0); + expect(finishChunkIndex).toBeGreaterThan(toolChunkIndex); +}); + +test("responsesSseToChatCompletionsSse keeps buffered arguments when the done snapshot is empty", async () => { + const frames = [ + `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: "" } })}\n\n`, + `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '{"cmd":' })}\n\n`, + `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '"pwd"}' })}\n\n`, + `event: response.output_item.done\ndata: ${JSON.stringify({ type: "response.output_item.done", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: "" } })}\n\n`, + `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { status: "completed" } })}\n\n`, + ]; + const { toolCalls } = await convertResponsesFrames(frames); + expect(toolCalls).toEqual([{ + index: 0, + id: "call_a", + type: "function", + function: { name: "exec_command", arguments: '{"cmd":"pwd"}' }, + }]); +}); + +test("responsesSseToChatCompletionsSse ignores duplicate done events for a tool call", async () => { + const done = `event: response.output_item.done\ndata: ${JSON.stringify({ type: "response.output_item.done", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: '{"cmd":"pwd"}' } })}\n\n`; + const frames = [ + `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: "" } })}\n\n`, + done, + done, + `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { status: "completed" } })}\n\n`, + ]; + const { toolCalls } = await convertResponsesFrames(frames); + expect(toolCalls).toEqual([{ + index: 0, + id: "call_a", + type: "function", + function: { name: "exec_command", arguments: '{"cmd":"pwd"}' }, + }]); +}); + +test("responsesSseToChatCompletionsSse uses finalized arguments when the item done event is absent", async () => { + const frames = [ + `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", id: "fc_a", call_id: "call_a", name: "exec_command", arguments: "" } })}\n\n`, + `event: response.function_call_arguments.delta\ndata: ${JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_a", delta: '{"cmd":"partial' })}\n\n`, + `event: response.function_call_arguments.done\ndata: ${JSON.stringify({ type: "response.function_call_arguments.done", item_id: "fc_a", name: "exec_command", arguments: '{"cmd":"final"}' })}\n\n`, + `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { status: "completed" } })}\n\n`, + ]; + const { chunks, toolCalls } = await convertResponsesFrames(frames); + expect(toolCalls).toEqual([{ + index: 0, + id: "call_a", + type: "function", + function: { name: "exec_command", arguments: '{"cmd":"final"}' }, + }]); + expect(chunks.some(chunk => chunk.choices?.[0]?.finish_reason === "tool_calls")).toBe(true); +}); + test("chatCompletionsToResponsesBody recovers tool_calls function.name from earlier call_id", () => { // Simulate replace-style client history: a later assistant tool_call has id+args but empty name, // while an earlier function_call in the same transcript already named it.