From f148c02c45b5710fa4c94a2cf5a28807009bbb1e Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Wed, 21 Oct 2026 09:45:42 -0400 Subject: [PATCH 1/4] fix(server-utils): Keep OpenAI streamed tool calls that have no index Co-Authored-By: Claude Sonnet 5.5 --- .../server-utils/src/ai/openai/streaming.ts | 32 ++++++- .../openai-streaming-tool-calls.test.ts | 94 +++++++++++++++++++ 2 files changed, 123 insertions(+), 3 deletions(-) create mode 100644 packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts diff --git a/packages/server-utils/src/ai/openai/streaming.ts b/packages/server-utils/src/ai/openai/streaming.ts index 0828e755e4cb..c706111fce97 100644 --- a/packages/server-utils/src/ai/openai/streaming.ts +++ b/packages/server-utils/src/ai/openai/streaming.ts @@ -37,7 +37,9 @@ interface StreamingState { * Accumulated tool calls from Chat Completion streaming, indexed by tool call index. * @see https://platform.openai.com/docs/guides/function-calling?api-mode=chat#streaming */ - chatCompletionToolCalls: Record; + chatCompletionToolCalls: Record; + /** Key of the latest tool call, which unindexed argument fragments continue. */ + lastChatCompletionToolCallKey?: number | string; /** * Accumulated function calls from Responses API streaming. * @see https://platform.openai.com/docs/guides/function-calling?api-mode=responses#streaming @@ -56,8 +58,8 @@ interface StreamingState { */ function processChatCompletionToolCalls(toolCalls: ChatCompletionToolCall[], state: StreamingState): void { for (const toolCall of toolCalls) { - const index = toolCall.index; - if (index === undefined || !toolCall.function) continue; + if (!toolCall.function) continue; + const index = getToolCallKey(toolCall, state); // Initialize tool call if this is the first chunk for this index if (!(index in state.chatCompletionToolCalls)) { @@ -75,9 +77,33 @@ function processChatCompletionToolCalls(toolCalls: ChatCompletionToolCall[], sta existingToolCall.function.arguments += toolCall.function.arguments; } } + state.lastChatCompletionToolCallKey = index; } } +/** + * OpenAI-compatible providers such as Gemini's OpenAI endpoint send tool calls without an `index`. + * Those are matched by a known `id`; a new `id` or a function name opens a new call, and anything + * else continues the latest call. + */ +function getToolCallKey(toolCall: ChatCompletionToolCall, state: StreamingState): number | string { + if (toolCall.index !== undefined) { + return toolCall.index; + } + if (toolCall.id) { + const known = Object.keys(state.chatCompletionToolCalls).find( + key => state.chatCompletionToolCalls[key]?.id === toolCall.id, + ); + if (known !== undefined) { + return known; + } + } + if (toolCall.id || toolCall.function?.name || state.lastChatCompletionToolCallKey === undefined) { + return `no-index-${Object.keys(state.chatCompletionToolCalls).length}`; + } + return state.lastChatCompletionToolCallKey; +} + /** * Processes a single OpenAI ChatCompletionChunk event, updating the streaming state. * diff --git a/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts new file mode 100644 index 000000000000..634b36e3a413 --- /dev/null +++ b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from 'vitest'; +import type { Span } from '@sentry/core'; +import { GEN_AI_RESPONSE_TOOL_CALLS } from '@sentry/conventions/attributes'; +import { instrumentStream } from '../../../../src/ai/openai/streaming'; +import type { ChatCompletionChunk, ChatCompletionToolCall } from '../../../../src/ai/openai/types'; + +function createMockSpan(): { span: Span; attributes: Record } { + const attributes: Record = {}; + let isEnded = false; + const span = { + isRecording: () => !isEnded, + setAttribute: (key: string, value: unknown) => { + attributes[key] = value; + }, + setAttributes: (attrs: Record) => { + Object.assign(attributes, attrs); + }, + setStatus: () => {}, + end: () => { + isEnded = true; + }, + } as unknown as Span; + return { span, attributes }; +} + +function chunkWith(toolCalls: Array>): ChatCompletionChunk { + return { + id: 'chatcmpl-1', + object: 'chat.completion.chunk', + model: 'gemini-test', + choices: [{ index: 0, delta: { tool_calls: toolCalls } }], + } as unknown as ChatCompletionChunk; +} + +async function recordedToolCalls(chunks: ChatCompletionChunk[]): Promise>> { + const { span, attributes } = createMockSpan(); + async function* stream(): AsyncGenerator { + yield* chunks; + } + for await (const _ of instrumentStream(stream(), span, true)) { + void _; + } + return JSON.parse(attributes[GEN_AI_RESPONSE_TOOL_CALLS] as string) as Array>; +} + +describe('instrumentStream (openai chat completion tool calls)', () => { + it('keeps whole tool calls that carry an id and no index', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([ + { id: 'call_a', type: 'function', function: { name: 'lookup', arguments: '{"q":"a"}' } }, + { id: 'call_b', type: 'function', function: { name: 'search', arguments: '{"q":"b"}' } }, + ]), + ]); + + expect(toolCalls).toEqual([ + { id: 'call_a', type: 'function', function: { name: 'lookup', arguments: '{"q":"a"}' } }, + { id: 'call_b', type: 'function', function: { name: 'search', arguments: '{"q":"b"}' } }, + ]); + }); + + it('joins argument fragments by id, then onto the latest call when there is no id', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([{ id: 'call_a', type: 'function', function: { name: 'lookup', arguments: '{"q":' } }]), + chunkWith([{ id: 'call_a', function: { name: '', arguments: '"a"' } }]), + chunkWith([{ function: { name: '', arguments: '}' } }]), + ]); + + expect(toolCalls).toEqual([ + { id: 'call_a', type: 'function', function: { name: 'lookup', arguments: '{"q":"a"}' } }, + ]); + }); + + it('keeps named calls that have neither id nor index', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([{ function: { name: 'lookup', arguments: '{}' } }]), + chunkWith([{ function: { name: 'search', arguments: '{}' } }]), + ]); + + expect(toolCalls.map(call => (call.function as { name: string }).name)).toEqual(['lookup', 'search']); + }); + + it('still groups indexed deltas by index', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([{ index: 1, id: 'call_b', function: { name: 'search', arguments: '{' } }]), + chunkWith([{ index: 0, id: 'call_a', function: { name: 'lookup', arguments: '{}' } }]), + chunkWith([{ index: 1, function: { name: '', arguments: '}' } }]), + ]); + + expect(toolCalls).toEqual([ + expect.objectContaining({ id: 'call_a', function: { name: 'lookup', arguments: '{}' } }), + expect.objectContaining({ id: 'call_b', function: { name: 'search', arguments: '{}' } }), + ]); + }); +}); From f0ac195ad54f7262880bd7f592d1c03362e1bb22 Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Wed, 21 Oct 2026 10:36:30 -0400 Subject: [PATCH 2/4] test(server-utils): cover an index-less OpenAI tool call followed by an indexed one Co-Authored-By: Claude Sonnet 5.5 --- .../ai/lib/tracing/openai-streaming-tool-calls.test.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts index 634b36e3a413..f68a09e41546 100644 --- a/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts @@ -79,6 +79,15 @@ describe('instrumentStream (openai chat completion tool calls)', () => { expect(toolCalls.map(call => (call.function as { name: string }).name)).toEqual(['lookup', 'search']); }); + it('does not let a later real index overwrite an earlier call without one', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([{ id: 'call_b', type: 'function', function: { name: 'search', arguments: '{}' } }]), + chunkWith([{ index: 0, id: 'call_a', function: { name: 'lookup', arguments: '{}' } }]), + ]); + + expect(toolCalls.map(call => call.id).sort()).toEqual(['call_a', 'call_b']); + }); + it('still groups indexed deltas by index', async () => { const toolCalls = await recordedToolCalls([ chunkWith([{ index: 1, id: 'call_b', function: { name: 'search', arguments: '{' } }]), From 349e0436c272f03b1e238157a7ff66a9c94efada Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Mon, 26 Oct 2026 15:46:45 -0400 Subject: [PATCH 3/4] fix(server-utils): Handle null and reused indexes in OpenAI streamed tool calls Co-Authored-By: Claude Sonnet 5.5 --- .../openai/openai-tool-calls/scenario.mjs | 44 +++++++++++++++++ .../tracing/openai/openai-tool-calls/test.ts | 38 ++++++++++++++- .../server-utils/src/ai/openai/streaming.ts | 31 ++++++++---- packages/server-utils/src/ai/openai/types.ts | 2 +- .../openai-streaming-tool-calls.test.ts | 47 +++++++++++++++++++ 5 files changed, 150 insertions(+), 12 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/scenario.mjs b/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/scenario.mjs index df34763d12c3..bb84084ecfa1 100644 --- a/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/scenario.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/scenario.mjs @@ -12,6 +12,9 @@ class MockOpenAIToolCalls { // If stream is requested, return an async generator if (params.stream) { + if (params.model === 'gemini-compat') { + return this._createGeminiStyleToolCallsStream(params); + } return this._createChatCompletionToolCallsStream(params); } @@ -139,6 +142,38 @@ class MockOpenAIToolCalls { }; } + // Gemini's OpenAI-compatible endpoint streams whole, parallel tool calls without an `index` + async *_createGeminiStyleToolCallsStream(params) { + yield { + id: 'chatcmpl-stream-gemini-tools-123', + object: 'chat.completion.chunk', + created: 1677652310, + model: params.model, + choices: [ + { + index: 0, + delta: { + role: 'assistant', + tool_calls: [ + { + id: 'call_paris', + type: 'function', + function: { name: 'get_weather', arguments: '{"latitude":48.8566,"longitude":2.3522}' }, + }, + { + id: 'call_london', + type: 'function', + function: { name: 'get_weather', arguments: '{"latitude":51.5072,"longitude":-0.1276}' }, + }, + ], + }, + finish_reason: 'tool_calls', + }, + ], + usage: { prompt_tokens: 15, completion_tokens: 25, total_tokens: 40 }, + }; + } + // Create a mock streaming response for responses API with tool calls async *_createResponsesApiToolCallsStream(params) { const responseId = 'resp_stream_tools_789'; @@ -275,6 +310,15 @@ async function run() { }); for await (const chunk of stream1) void chunk; + // Test 2b: Chat completion streaming whole parallel tool calls without an index (Gemini style) + const geminiStream = await client.chat.completions.create({ + model: 'gemini-compat', + messages: [message], + tools: [weatherTool], + stream: true, + }); + for await (const chunk of geminiStream) void chunk; + // Test 3: Responses API with tools (non-streaming) await client.responses.create({ model: 'gpt-4', diff --git a/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/test.ts b/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/test.ts index 59473fcea8fe..cf5e050605da 100644 --- a/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/openai/openai-tool-calls/test.ts @@ -66,6 +66,25 @@ describe('OpenAI Tool Calls integration', () => { }, ]); + const CHAT_GEMINI_STREAM_TOOL_CALLS = JSON.stringify([ + { + id: 'call_paris', + type: 'function', + function: { + name: 'get_weather', + arguments: '{"latitude":48.8566,"longitude":2.3522}', + }, + }, + { + id: 'call_london', + type: 'function', + function: { + name: 'get_weather', + arguments: '{"latitude":51.5072,"longitude":-0.1276}', + }, + }, + ]); + const RESPONSES_TOOL_CALLS = JSON.stringify([ { type: 'function_call', @@ -84,7 +103,7 @@ describe('OpenAI Tool Calls integration', () => { const segment = container.items.find(span => span.is_segment && span.name === 'main'); expect(segment).toBeDefined(); const spans = container.items.filter(span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.openai'); - expect(spans).toHaveLength(4); + expect(spans).toHaveLength(5); const chatToolsSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-tools-123', ); @@ -193,6 +212,12 @@ describe('OpenAI Tool Calls integration', () => { value: 40, }); + const geminiStreamSpan = spans.find( + span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-gemini-tools-123', + ); + expect(geminiStreamSpan).toBeDefined(); + expect(geminiStreamSpan!.attributes[GEN_AI_RESPONSE_TOOL_CALLS]).toBeUndefined(); + const responsesToolsSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'resp_tools_789', ); @@ -318,7 +343,7 @@ describe('OpenAI Tool Calls integration', () => { const segment = container.items.find(span => span.is_segment && span.name === 'main'); expect(segment).toBeDefined(); const spans = container.items.filter(span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.openai'); - expect(spans).toHaveLength(4); + expect(spans).toHaveLength(5); const chatToolsSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-tools-123', ); @@ -452,6 +477,15 @@ describe('OpenAI Tool Calls integration', () => { value: 40, }); + const geminiStreamSpan = spans.find( + span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-gemini-tools-123', + ); + expect(geminiStreamSpan).toBeDefined(); + expect(geminiStreamSpan!.attributes[GEN_AI_RESPONSE_TOOL_CALLS]).toEqual({ + type: 'string', + value: CHAT_GEMINI_STREAM_TOOL_CALLS, + }); + const responsesToolsSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'resp_tools_789', ); diff --git a/packages/server-utils/src/ai/openai/streaming.ts b/packages/server-utils/src/ai/openai/streaming.ts index c706111fce97..a1adc703f130 100644 --- a/packages/server-utils/src/ai/openai/streaming.ts +++ b/packages/server-utils/src/ai/openai/streaming.ts @@ -40,6 +40,8 @@ interface StreamingState { chatCompletionToolCalls: Record; /** Key of the latest tool call, which unindexed argument fragments continue. */ lastChatCompletionToolCallKey?: number | string; + /** Key of the call currently open for a reported `index`, when a new `id` reused that index. */ + chatCompletionToolCallKeyByIndex?: Record; /** * Accumulated function calls from Responses API streaming. * @see https://platform.openai.com/docs/guides/function-calling?api-mode=responses#streaming @@ -82,24 +84,35 @@ function processChatCompletionToolCalls(toolCalls: ChatCompletionToolCall[], sta } /** - * OpenAI-compatible providers such as Gemini's OpenAI endpoint send tool calls without an `index`. - * Those are matched by a known `id`; a new `id` or a function name opens a new call, and anything - * else continues the latest call. + * OpenAI-compatible providers such as Gemini's OpenAI endpoint send tool calls without an `index` + * (or with `null`), or reuse the same `index` for every call. A call without a usable `index` is + * matched by a known `id`; a new `id` or a function name opens a new call, and anything else + * continues the latest call. A different non-empty `id` on an `index` that is already taken also + * opens a new call. */ function getToolCallKey(toolCall: ChatCompletionToolCall, state: StreamingState): number | string { - if (toolCall.index !== undefined) { - return toolCall.index; + const calls = state.chatCompletionToolCalls; + const findById = (id: string): string | undefined => Object.keys(calls).find(key => calls[key]?.id === id); + + if (typeof toolCall.index === 'number') { + const keyByIndex = (state.chatCompletionToolCallKeyByIndex ??= {}); + const key = keyByIndex[toolCall.index] ?? toolCall.index; + const existingId = calls[key]?.id; + if (existingId && toolCall.id && existingId !== toolCall.id) { + const next = findById(toolCall.id) ?? `index-${toolCall.index}-${Object.keys(calls).length}`; + keyByIndex[toolCall.index] = next; + return next; + } + return key; } if (toolCall.id) { - const known = Object.keys(state.chatCompletionToolCalls).find( - key => state.chatCompletionToolCalls[key]?.id === toolCall.id, - ); + const known = findById(toolCall.id); if (known !== undefined) { return known; } } if (toolCall.id || toolCall.function?.name || state.lastChatCompletionToolCallKey === undefined) { - return `no-index-${Object.keys(state.chatCompletionToolCalls).length}`; + return `no-index-${Object.keys(calls).length}`; } return state.lastChatCompletionToolCallKey; } diff --git a/packages/server-utils/src/ai/openai/types.ts b/packages/server-utils/src/ai/openai/types.ts index cb57ea84b7a0..43bf03e51288 100644 --- a/packages/server-utils/src/ai/openai/types.ts +++ b/packages/server-utils/src/ai/openai/types.ts @@ -176,7 +176,7 @@ interface ResponseOutputItemDoneEvent { * Tool call object for Chat Completion streaming */ export interface ChatCompletionToolCall { - index?: number; // Present for streaming responses + index?: number | null; // Present for streaming responses; some compatible providers send null id: string; type?: string; // Could be missing for streaming responses function?: { diff --git a/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts index f68a09e41546..7ddbdfc849b4 100644 --- a/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts @@ -100,4 +100,51 @@ describe('instrumentStream (openai chat completion tool calls)', () => { expect.objectContaining({ id: 'call_b', function: { name: 'search', arguments: '{}' } }), ]); }); + + it('treats a null index like a missing one', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([ + { index: null, id: 'call_a', type: 'function', function: { name: 'lookup', arguments: '{"q":"a"}' } }, + { index: null, id: 'call_b', type: 'function', function: { name: 'search', arguments: '{"q":"b"}' } }, + ]), + chunkWith([{ index: null, function: { name: '', arguments: ' ' } }]), + ]); + + expect(toolCalls.map(call => call.id)).toEqual(['call_a', 'call_b']); + expect((toolCalls[1]?.function as { arguments: string }).arguments).toBe('{"q":"b"} '); + }); + + it('opens a new call when a different id lands on an index that is already taken', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([ + { index: 0, id: 'call_a', type: 'function', function: { name: 'lookup', arguments: '{"q":"a"}' } }, + { index: 0, id: 'call_b', type: 'function', function: { name: 'search', arguments: '{"q":"b"}' } }, + ]), + ]); + + expect(toolCalls).toEqual([ + { index: 0, id: 'call_a', type: 'function', function: { name: 'lookup', arguments: '{"q":"a"}' } }, + { index: 0, id: 'call_b', type: 'function', function: { name: 'search', arguments: '{"q":"b"}' } }, + ]); + }); + + it('continues the latest call opened on a reused index when the next delta has no id', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([{ index: 0, id: 'call_a', function: { name: 'lookup', arguments: '{}' } }]), + chunkWith([{ index: 0, id: 'call_b', function: { name: 'search', arguments: '{"q":' } }]), + chunkWith([{ index: 0, function: { name: '', arguments: '"b"}' } }]), + ]); + + expect(toolCalls.map(call => (call.function as { arguments: string }).arguments)).toEqual(['{}', '{"q":"b"}']); + }); + + it('does not split a call whose id is repeated on every delta of its index', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([{ index: 0, id: 'call_a', function: { name: 'lookup', arguments: '{"q":' } }]), + chunkWith([{ index: 0, id: 'call_a', function: { name: '', arguments: '"a"}' } }]), + ]); + + expect(toolCalls).toHaveLength(1); + expect((toolCalls[0]?.function as { arguments: string }).arguments).toBe('{"q":"a"}'); + }); }); From 0d1c6f76c189b8b6b458ccd5e7c91a280b16c32b Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Wed, 28 Oct 2026 05:32:09 -0400 Subject: [PATCH 4/4] fix(server-utils): Keep OpenAI streamed tool calls in the order they were opened Co-Authored-By: Claude Sonnet 5.5 --- .../suites/tracing/openai/scenario-chat.mjs | 92 ++++++++++++++----- .../suites/tracing/openai/test.ts | 37 +++++++- .../server-utils/src/ai/openai/streaming.ts | 43 +++++---- .../openai-streaming-tool-calls.test.ts | 16 +++- 4 files changed, 143 insertions(+), 45 deletions(-) diff --git a/dev-packages/node-integration-tests/suites/tracing/openai/scenario-chat.mjs b/dev-packages/node-integration-tests/suites/tracing/openai/scenario-chat.mjs index 6b8bc76b582b..84e2c2a0bddc 100644 --- a/dev-packages/node-integration-tests/suites/tracing/openai/scenario-chat.mjs +++ b/dev-packages/node-integration-tests/suites/tracing/openai/scenario-chat.mjs @@ -22,35 +22,74 @@ function startMockServer() { res.setHeader('Cache-Control', 'no-cache'); res.setHeader('Connection', 'keep-alive'); - const chunks = [ + // Gemini's OpenAI-compatible endpoint streams whole, parallel tool calls without an `index` + const geminiToolCallChunks = [ { - id: 'chatcmpl-stream-123', + id: 'chatcmpl-stream-gemini-456', object: 'chat.completion.chunk', - created: 1677652300, + created: 1677652305, model: model, - choices: [{ delta: { role: 'assistant', content: '' }, index: 0 }], - }, - { - id: 'chatcmpl-stream-123', - object: 'chat.completion.chunk', - created: 1677652300, - model: model, - choices: [{ delta: { content: 'Hello from OpenAI streaming!' }, index: 0 }], - }, - { - id: 'chatcmpl-stream-123', - object: 'chat.completion.chunk', - created: 1677652300, - model: model, - choices: [{ delta: {}, index: 0, finish_reason: 'stop' }], + choices: [ + { + index: 0, + delta: { + role: 'assistant', + tool_calls: [ + { + id: 'call_paris', + type: 'function', + function: { name: 'get_weather', arguments: '{"city":"Paris"}' }, + }, + { + id: 'call_london', + type: 'function', + function: { name: 'get_weather', arguments: '{"city":"London"}' }, + }, + ], + }, + finish_reason: 'tool_calls', + }, + ], usage: { - prompt_tokens: 12, - completion_tokens: 18, - total_tokens: 30, + prompt_tokens: 15, + completion_tokens: 25, + total_tokens: 40, }, }, ]; + const chunks = + model === 'gemini-compat' + ? geminiToolCallChunks + : [ + { + id: 'chatcmpl-stream-123', + object: 'chat.completion.chunk', + created: 1677652300, + model: model, + choices: [{ delta: { role: 'assistant', content: '' }, index: 0 }], + }, + { + id: 'chatcmpl-stream-123', + object: 'chat.completion.chunk', + created: 1677652300, + model: model, + choices: [{ delta: { content: 'Hello from OpenAI streaming!' }, index: 0 }], + }, + { + id: 'chatcmpl-stream-123', + object: 'chat.completion.chunk', + created: 1677652300, + model: model, + choices: [{ delta: {}, index: 0, finish_reason: 'stop' }], + usage: { + prompt_tokens: 12, + completion_tokens: 18, + total_tokens: 30, + }, + }, + ]; + chunks.forEach((chunk, index) => { setTimeout(() => { res.write(`data: ${JSON.stringify(chunk)}\n\n`); @@ -243,6 +282,17 @@ async function run() { void chunk; // Prevent unused variable warning } + // Streaming whole parallel tool calls without an index (Gemini style) + const geminiStream = await client.chat.completions.create({ + model: 'gemini-compat', + messages: [{ role: 'user', content: 'Weather in Paris and London?' }], + stream: true, + }); + + for await (const chunk of geminiStream) { + void chunk; + } + // Fifth test: responses API streaming const stream2 = await client.responses.create({ model: 'gpt-4', diff --git a/dev-packages/node-integration-tests/suites/tracing/openai/test.ts b/dev-packages/node-integration-tests/suites/tracing/openai/test.ts index 4db70ae10643..41b57d713565 100644 --- a/dev-packages/node-integration-tests/suites/tracing/openai/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/openai/test.ts @@ -13,6 +13,7 @@ import { GEN_AI_RESPONSE_MODEL, GEN_AI_RESPONSE_STREAMING, GEN_AI_RESPONSE_TEXT, + GEN_AI_RESPONSE_TOOL_CALLS, GEN_AI_SYSTEM_INSTRUCTIONS, GEN_AI_USAGE_INPUT_TOKENS, GEN_AI_USAGE_OUTPUT_TOKENS, @@ -41,7 +42,7 @@ describe('OpenAI integration', () => { const allSpans = await spansPromise; const spans = allSpans.filter(span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.openai'); - expect(spans).toHaveLength(8); + expect(spans).toHaveLength(9); const chatCompletionSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-mock123' && @@ -173,6 +174,12 @@ describe('OpenAI integration', () => { value: 'error-model', }); + const geminiStreamSpan = spans.find( + span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-gemini-456', + ); + expect(geminiStreamSpan).toBeDefined(); + expect(geminiStreamSpan!.attributes[GEN_AI_RESPONSE_TOOL_CALLS]).toBeUndefined(); + const streamingChatCompletionSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-123', ); @@ -444,7 +451,7 @@ describe('OpenAI integration', () => { const allSpans = await spansPromise; const spans = allSpans.filter(span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.openai'); - expect(spans).toHaveLength(8); + expect(spans).toHaveLength(9); const chatCompletionSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-mock123' && @@ -600,6 +607,18 @@ describe('OpenAI integration', () => { value: '[{"role":"user","content":"This will fail"}]', }); + const geminiStreamSpan = spans.find( + span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-gemini-456', + ); + expect(geminiStreamSpan).toBeDefined(); + expect(geminiStreamSpan!.attributes[GEN_AI_RESPONSE_TOOL_CALLS]).toEqual({ + type: 'string', + value: JSON.stringify([ + { id: 'call_paris', type: 'function', function: { name: 'get_weather', arguments: '{"city":"Paris"}' } }, + { id: 'call_london', type: 'function', function: { name: 'get_weather', arguments: '{"city":"London"}' } }, + ]), + }); + const streamingChatCompletionSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-123', ); @@ -902,7 +921,7 @@ describe('OpenAI integration', () => { const allSpans = await spansPromise; const spans = allSpans.filter(span => span.attributes[SENTRY_ORIGIN]?.value === 'auto.ai.openai'); - expect(spans).toHaveLength(8); + expect(spans).toHaveLength(9); const chatCompletionSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-mock123' && @@ -918,6 +937,18 @@ describe('OpenAI integration', () => { value: expect.any(String), }); + const geminiStreamSpan = spans.find( + span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-gemini-456', + ); + expect(geminiStreamSpan).toBeDefined(); + expect(geminiStreamSpan!.attributes[GEN_AI_RESPONSE_TOOL_CALLS]).toEqual({ + type: 'string', + value: JSON.stringify([ + { id: 'call_paris', type: 'function', function: { name: 'get_weather', arguments: '{"city":"Paris"}' } }, + { id: 'call_london', type: 'function', function: { name: 'get_weather', arguments: '{"city":"London"}' } }, + ]), + }); + const streamingChatCompletionSpan = spans.find( span => span.attributes[GEN_AI_RESPONSE_ID]?.value === 'chatcmpl-stream-123', ); diff --git a/packages/server-utils/src/ai/openai/streaming.ts b/packages/server-utils/src/ai/openai/streaming.ts index a1adc703f130..15686abfa072 100644 --- a/packages/server-utils/src/ai/openai/streaming.ts +++ b/packages/server-utils/src/ai/openai/streaming.ts @@ -34,14 +34,14 @@ interface StreamingState { cacheReadInputTokens: number | undefined; cacheCreationInputTokens: number | undefined; /** - * Accumulated tool calls from Chat Completion streaming, indexed by tool call index. + * Accumulated tool calls from Chat Completion streaming, keyed by string in the order they were opened. * @see https://platform.openai.com/docs/guides/function-calling?api-mode=chat#streaming */ - chatCompletionToolCalls: Record; + chatCompletionToolCalls: Map; /** Key of the latest tool call, which unindexed argument fragments continue. */ - lastChatCompletionToolCallKey?: number | string; + lastChatCompletionToolCallKey?: string; /** Key of the call currently open for a reported `index`, when a new `id` reused that index. */ - chatCompletionToolCallKeyByIndex?: Record; + chatCompletionToolCallKeyByIndex?: Map; /** * Accumulated function calls from Responses API streaming. * @see https://platform.openai.com/docs/guides/function-calling?api-mode=responses#streaming @@ -64,18 +64,18 @@ function processChatCompletionToolCalls(toolCalls: ChatCompletionToolCall[], sta const index = getToolCallKey(toolCall, state); // Initialize tool call if this is the first chunk for this index - if (!(index in state.chatCompletionToolCalls)) { - state.chatCompletionToolCalls[index] = { + const existingToolCall = state.chatCompletionToolCalls.get(index); + if (!existingToolCall) { + state.chatCompletionToolCalls.set(index, { ...toolCall, function: { name: toolCall.function.name, arguments: toolCall.function.arguments || '', }, - }; + }); } else { // Accumulate function arguments from subsequent chunks - const existingToolCall = state.chatCompletionToolCalls[index]; - if (toolCall.function.arguments && existingToolCall?.function) { + if (toolCall.function.arguments && existingToolCall.function) { existingToolCall.function.arguments += toolCall.function.arguments; } } @@ -90,17 +90,22 @@ function processChatCompletionToolCalls(toolCalls: ChatCompletionToolCall[], sta * continues the latest call. A different non-empty `id` on an `index` that is already taken also * opens a new call. */ -function getToolCallKey(toolCall: ChatCompletionToolCall, state: StreamingState): number | string { +function getToolCallKey(toolCall: ChatCompletionToolCall, state: StreamingState): string { const calls = state.chatCompletionToolCalls; - const findById = (id: string): string | undefined => Object.keys(calls).find(key => calls[key]?.id === id); + const findById = (id: string): string | undefined => { + for (const [key, call] of calls) { + if (call.id === id) return key; + } + return undefined; + }; if (typeof toolCall.index === 'number') { - const keyByIndex = (state.chatCompletionToolCallKeyByIndex ??= {}); - const key = keyByIndex[toolCall.index] ?? toolCall.index; - const existingId = calls[key]?.id; + const keyByIndex = (state.chatCompletionToolCallKeyByIndex ??= new Map()); + const key = keyByIndex.get(toolCall.index) ?? `index-${toolCall.index}`; + const existingId = calls.get(key)?.id; if (existingId && toolCall.id && existingId !== toolCall.id) { - const next = findById(toolCall.id) ?? `index-${toolCall.index}-${Object.keys(calls).length}`; - keyByIndex[toolCall.index] = next; + const next = findById(toolCall.id) ?? `index-${toolCall.index}-${calls.size}`; + keyByIndex.set(toolCall.index, next); return next; } return key; @@ -112,7 +117,7 @@ function getToolCallKey(toolCall: ChatCompletionToolCall, state: StreamingState) } } if (toolCall.id || toolCall.function?.name || state.lastChatCompletionToolCallKey === undefined) { - return `no-index-${Object.keys(calls).length}`; + return `no-index-${calls.size}`; } return state.lastChatCompletionToolCallKey; } @@ -264,7 +269,7 @@ export async function* instrumentStream( totalTokens: undefined, cacheReadInputTokens: undefined, cacheCreationInputTokens: undefined, - chatCompletionToolCalls: {}, + chatCompletionToolCalls: new Map(), responsesApiToolCalls: [], }; @@ -278,7 +283,7 @@ export async function* instrumentStream( yield event; } } finally { - const allToolCalls = [...Object.values(state.chatCompletionToolCalls), ...state.responsesApiToolCalls]; + const allToolCalls = [...state.chatCompletionToolCalls.values(), ...state.responsesApiToolCalls]; endStreamSpan(span, { ...state, toolCalls: allToolCalls }, recordOutputs); } } diff --git a/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts index 7ddbdfc849b4..cd39e060c9bc 100644 --- a/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts +++ b/packages/server-utils/test/ai/lib/tracing/openai-streaming-tool-calls.test.ts @@ -85,13 +85,13 @@ describe('instrumentStream (openai chat completion tool calls)', () => { chunkWith([{ index: 0, id: 'call_a', function: { name: 'lookup', arguments: '{}' } }]), ]); - expect(toolCalls.map(call => call.id).sort()).toEqual(['call_a', 'call_b']); + expect(toolCalls.map(call => call.id)).toEqual(['call_b', 'call_a']); }); it('still groups indexed deltas by index', async () => { const toolCalls = await recordedToolCalls([ - chunkWith([{ index: 1, id: 'call_b', function: { name: 'search', arguments: '{' } }]), chunkWith([{ index: 0, id: 'call_a', function: { name: 'lookup', arguments: '{}' } }]), + chunkWith([{ index: 1, id: 'call_b', function: { name: 'search', arguments: '{' } }]), chunkWith([{ index: 1, function: { name: '', arguments: '}' } }]), ]); @@ -147,4 +147,16 @@ describe('instrumentStream (openai chat completion tool calls)', () => { expect(toolCalls).toHaveLength(1); expect((toolCalls[0]?.function as { arguments: string }).arguments).toBe('{"q":"a"}'); }); + + it('keeps the order calls were opened when a reused index is followed by a higher index', async () => { + const toolCalls = await recordedToolCalls([ + chunkWith([ + { index: 0, id: 'call_a', function: { name: 'lookup', arguments: '{}' } }, + { index: 0, id: 'call_b', function: { name: 'search', arguments: '{}' } }, + { index: 1, id: 'call_c', function: { name: 'fetch', arguments: '{}' } }, + ]), + ]); + + expect(toolCalls.map(call => call.id)).toEqual(['call_a', 'call_b', 'call_c']); + }); });