From 0f7c2ae008f2b94b3e35dbfd0fe6919a9db24629 Mon Sep 17 00:00:00 2001 From: Dave Page Date: Wed, 26 Aug 2026 14:05:08 +0100 Subject: [PATCH] Fix OpenAI streaming tool calls being split across two entries _read_responses_stream() keyed its tool_calls_data dict by call_id, but response.function_call_arguments.delta events carry item_id, not call_id, so the delta accumulation always missed the entry seeded by the preceding response.output_item.added event and created a second, empty-name entry instead. Correlate on item_id, keeping the real call_id alongside it so the emitted ToolCall.id is still correct. Closes #10348 --- web/pgadmin/llm/providers/openai.py | 24 ++++---- web/pgadmin/llm/tests/test_openai_stream.py | 61 +++++++++++++++++++++ 2 files changed, 75 insertions(+), 10 deletions(-) diff --git a/web/pgadmin/llm/providers/openai.py b/web/pgadmin/llm/providers/openai.py index c55ee642bf4..a7ff2778379 100644 --- a/web/pgadmin/llm/providers/openai.py +++ b/web/pgadmin/llm/providers/openai.py @@ -844,7 +844,10 @@ def _read_responses_stream( response.completed for the final response. """ content_parts = [] - # tool_calls_data: {call_id: {name, arguments}} + # tool_calls_data: {item_id: {call_id, name, arguments}} + # Keyed by item_id because response.function_call_arguments.delta + # events carry item_id (matching item.id from output_item.added), + # not call_id. tool_calls_data = {} model_name = self._model usage = Usage() @@ -884,19 +887,20 @@ def _read_responses_stream( elif event_type == 'response.output_item.added': item = data.get('item', {}) if item.get('type') == 'function_call': - call_id = item.get('call_id', '') - tool_calls_data[call_id] = { + item_id = item.get('id', '') + tool_calls_data[item_id] = { + 'call_id': item.get('call_id', ''), 'name': item.get('name', ''), 'arguments': '' } elif event_type == 'response.function_call_arguments.delta': - call_id = data.get('call_id', '') - if call_id not in tool_calls_data: - tool_calls_data[call_id] = { - 'name': '', 'arguments': '' + item_id = data.get('item_id', '') + if item_id not in tool_calls_data: + tool_calls_data[item_id] = { + 'call_id': '', 'name': '', 'arguments': '' } - tool_calls_data[call_id]['arguments'] += data.get( + tool_calls_data[item_id]['arguments'] += data.get( 'delta', '' ) @@ -915,14 +919,14 @@ def _read_responses_stream( # Build final response content = ''.join(content_parts) tool_calls = [] - for call_id, tc in tool_calls_data.items(): + for tc in tool_calls_data.values(): try: arguments = json.loads(tc['arguments']) \ if tc['arguments'] else {} except json.JSONDecodeError: arguments = {} tool_calls.append(ToolCall( - id=call_id or str(uuid.uuid4()), + id=tc['call_id'] or str(uuid.uuid4()), name=tc['name'], arguments=arguments )) diff --git a/web/pgadmin/llm/tests/test_openai_stream.py b/web/pgadmin/llm/tests/test_openai_stream.py index 75e36aed06b..59b97afb9da 100644 --- a/web/pgadmin/llm/tests/test_openai_stream.py +++ b/web/pgadmin/llm/tests/test_openai_stream.py @@ -118,3 +118,64 @@ def runTest(self): # The real provider id must survive a null id in a later delta, # rather than being clobbered (and replaced by a random uuid). self.assertEqual(tc.id, self.expected_id) + + +class OpenAIResponsesStreamToolCallTestCase(BaseTestGenerator): + """Responses API function-call deltas must be correlated on item_id, + not call_id (issue #10348): response.function_call_arguments.delta + events carry item_id, not call_id. + """ + + scenarios = [ + ('A single streamed tool call keeps its name and arguments ' + 'together', dict( + stream=[ + _sse({'type': 'response.output_item.added', 'item': { + 'type': 'function_call', 'id': 'item_1', + 'call_id': 'call_abc', 'name': 'get_database_schema' + }}), + _sse({'type': 'response.function_call_arguments.delta', + 'item_id': 'item_1', 'delta': '{"table":'}), + _sse({'type': 'response.function_call_arguments.delta', + 'item_id': 'item_1', 'delta': '"users"}'}), + _sse({'type': 'response.completed', 'response': {}}), + ], + expected=[ + ('call_abc', 'get_database_schema', {'table': 'users'}), + ], + )), + ('Two parallel tool calls are not merged into one', dict( + stream=[ + _sse({'type': 'response.output_item.added', 'item': { + 'type': 'function_call', 'id': 'item_1', + 'call_id': 'call_1', 'name': 'run_query' + }}), + _sse({'type': 'response.output_item.added', 'item': { + 'type': 'function_call', 'id': 'item_2', + 'call_id': 'call_2', 'name': 'get_database_schema' + }}), + _sse({'type': 'response.function_call_arguments.delta', + 'item_id': 'item_1', 'delta': '{"sql": "SELECT 1"}'}), + _sse({'type': 'response.function_call_arguments.delta', + 'item_id': 'item_2', 'delta': '{}'}), + _sse({'type': 'response.completed', 'response': {}}), + ], + expected=[ + ('call_1', 'run_query', {'sql': 'SELECT 1'}), + ('call_2', 'get_database_schema', {}), + ], + )), + ] + + def runTest(self): + client = OpenAIClient(api_key='test-key', model='gpt-5') + result = None + for item in client._read_responses_stream(_FakeStream(self.stream)): + if isinstance(item, LLMResponse): + result = item + + self.assertIsNotNone(result) + self.assertEqual(len(result.tool_calls), len(self.expected)) + actual = [(tc.id, tc.name, tc.arguments) + for tc in result.tool_calls] + self.assertEqual(actual, self.expected)