From f9f96c58c00b767a7ca3475526a7fc6c56803490 Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 15:16:47 -0400 Subject: [PATCH 01/10] fix(responses): preserve streamed function call identity --- .../streaming_iterator.py | 82 ++++- .../test_streaming_iterator_transformation.py | 334 +++++++++++++++++- 2 files changed, 409 insertions(+), 7 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index b2edf2bf9ed..d389488e71a 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -114,7 +114,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._pending_tool_events: list[BaseLiteLLMOpenAIResponseObject] = [] self._tool_output_index_by_call_id: dict[str, int] = {} self._tool_args_by_call_id: dict[str, str] = {} + self._tool_item_id_by_call_id: dict[str, str] = {} self._tool_call_id_by_index: dict[int, str] = {} + self._streamed_tool_call_ids_in_order: list[str] = [] + self._resolved_tool_call_id_by_position: dict[int, str] = {} self._ambiguous_tool_call_indexes: set[int] = set() self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item self._final_tool_events_queued: bool = False @@ -153,6 +156,34 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): except (TypeError, ValueError): return None + def _streamed_tool_call_id_at_position(self, position: int) -> str | None: + if position in self._ambiguous_tool_call_indexes: + return None + indexed_call_id: Final = self._tool_call_id_by_index.get(position) + if indexed_call_id is not None: + return indexed_call_id + # If the stream supplied any indexes, a missing position is a terminal-only + # call. Falling back to arrival order here could conflate parallel calls. + if self._tool_call_id_by_index: + return None + streamed_call_ids: Final = getattr(self, "_streamed_tool_call_ids_in_order", ()) + if position < len(streamed_call_ids): + return streamed_call_ids[position] + return None + + def _streamed_tool_call_id_for_terminal_call(self, tool_call: object, position: int) -> str | None: + """Match a terminal aggregate tool call to the identity emitted while streaming.""" + tool_call_index: Final = self._normalize_tool_call_index(tool_call) + if tool_call_index is not None: + if tool_call_index in self._ambiguous_tool_call_indexes: + return None + indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) + if indexed_call_id is not None: + return indexed_call_id + if self._tool_call_id_by_index: + return None + return self._streamed_tool_call_id_at_position(position) + def _responses_namespace_tool_call_fields(self, fn_name: str) -> tuple[str, str | None]: mapped: Final = self._namespace_tool_names.get(fn_name) if mapped: @@ -224,9 +255,17 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if call_id not in self._tool_args_by_call_id: self._tool_args_by_call_id[call_id] = "" + streamed_call_ids = getattr(self, "_streamed_tool_call_ids_in_order", None) + if streamed_call_ids is None: + streamed_call_ids = self._streamed_tool_call_ids_in_order = [] + streamed_call_ids.append(call_id) self._sequence_number += 1 names = self._custom_tool_names item_kwargs = build_tool_call_item_kwargs(call_id, tool_name, "", "in_progress", names) + tool_item_ids = getattr(self, "_tool_item_id_by_call_id", None) + if tool_item_ids is None: + tool_item_ids = self._tool_item_id_by_call_id = {} + tool_item_ids[call_id] = item_kwargs["id"] if tool_namespace: item_kwargs["namespace"] = tool_namespace event = OutputItemAddedEvent( @@ -248,7 +287,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._sequence_number += 1 delta_event: BaseLiteLLMOpenAIResponseObject = FunctionCallArgumentsDeltaEvent( type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, - item_id=call_id, + item_id=self._tool_item_id_by_call_id.get(call_id, call_id), output_index=output_index, delta=delta_chunk, ) @@ -273,11 +312,15 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if not tool_calls or not isinstance(tool_calls, list): return - for tc in tool_calls: + for position, tc in enumerate(tool_calls): call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None) if not call_id_raw: continue - call_id = str(call_id_raw) + call_id = self._streamed_tool_call_id_for_terminal_call(tc, position) or str(call_id_raw) + resolved_call_ids = getattr(self, "_resolved_tool_call_id_by_position", None) + if resolved_call_ids is None: + resolved_call_ids = self._resolved_tool_call_id_by_position = {} + resolved_call_ids[position] = call_id output_index = self._get_or_assign_tool_output_index(call_id) fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) @@ -300,6 +343,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._sequence_number += 1 names = self._custom_tool_names item_kwargs = build_tool_call_item_kwargs(call_id, tool_name, "", "in_progress", names) + tool_item_ids = getattr(self, "_tool_item_id_by_call_id", None) + if tool_item_ids is None: + tool_item_ids = self._tool_item_id_by_call_id = {} + tool_item_ids[call_id] = item_kwargs["id"] if tool_namespace: item_kwargs["namespace"] = tool_namespace event = OutputItemAddedEvent( @@ -325,7 +372,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._sequence_number += 1 delta_event = FunctionCallArgumentsDeltaEvent( type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, - item_id=call_id, + item_id=self._tool_item_id_by_call_id.get(call_id, call_id), output_index=output_index, delta=delta_chunk, ) @@ -335,7 +382,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._sequence_number += 1 done_event = FunctionCallArgumentsDoneEvent( type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, - item_id=call_id, + item_id=self._tool_item_id_by_call_id.get(call_id, call_id), output_index=output_index, arguments=final_args, ) @@ -345,6 +392,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._sequence_number += 1 names = self._custom_tool_names item_kwargs = build_tool_call_item_kwargs(call_id, tool_name, final_args, "completed", names) + tool_item_ids = getattr(self, "_tool_item_id_by_call_id", None) + if tool_item_ids is None: + tool_item_ids = self._tool_item_id_by_call_id = {} + item_kwargs["id"] = tool_item_ids.setdefault(call_id, item_kwargs["id"]) if tool_namespace: item_kwargs["namespace"] = tool_namespace item_done_event = OutputItemDoneEvent( @@ -1161,7 +1212,26 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): "message", self._cached_item_id, ) - return _output_items_with_id(message_aligned, "reasoning", self._cached_reasoning_item_id) + reasoning_aligned: Final = _output_items_with_id( + message_aligned, + "reasoning", + self._cached_reasoning_item_id, + ) + tool_position = 0 + aligned_items: list[Any] = [] + for item in reasoning_aligned: + if getattr(item, "type", None) in {"function_call", "custom_tool_call"}: + resolved_call_ids: Final = getattr(self, "_resolved_tool_call_id_by_position", {}) + streamed_call_id = resolved_call_ids.get(tool_position) + if streamed_call_id is None: + streamed_call_id = self._streamed_tool_call_id_at_position(tool_position) + tool_position += 1 + if streamed_call_id is not None: + tool_item_ids: Final = getattr(self, "_tool_item_id_by_call_id", {}) + streamed_item_id = tool_item_ids.get(streamed_call_id, getattr(item, "id", streamed_call_id)) + item = item.model_copy(update={"id": streamed_item_id, "call_id": streamed_call_id}) + aligned_items.append(item) + return tuple(aligned_items) def _emit_response_completed_event(self, litellm_model_response: ModelResponse) -> ResponseCompletedEvent | None: if litellm_model_response: diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 01148f627f1..cd54accc616 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -11,7 +11,7 @@ spend tracking stores, so a follow-up previous_response_id still finds the conve """ import json -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -408,6 +408,338 @@ def test_parallel_tool_calls_without_ids_use_index_mapping(): assert arguments_by_call_id["call_b"] == '{"y":2}' +def test_final_tool_events_and_completed_snapshot_reuse_streamed_call_identity(): + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + streamed_ids = ["call_stream_a", "call_stream_b"] + terminal_ids = ["call_terminal_a", "call_terminal_b"] + + iterator._queue_tool_call_delta_events( + [ + { + "index": index, + "id": call_id, + "type": "function", + "function": { + "name": f"tool_{index}", + "arguments": f'{{"value":{index}', + }, + } + for index, call_id in reversed(list(enumerate(streamed_ids))) + ] + ) + # Simulate delivery of all incremental events before the terminal aggregate arrives. + iterator._pending_tool_events.clear() + + iterator.litellm_model_response = ModelResponse( + id="chatcmpl-terminal", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": terminal_id, + "type": "function", + "function": { + "name": f"tool_{index}", + "arguments": f'{{"value":{index}}}', + }, + } + for index, terminal_id in enumerate(terminal_ids) + ], + }, + } + ], + ) + + final_events = [] + for _ in range(20): + event = iterator.common_done_event_logic() + final_events.append(event) + if event.type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED: + break + else: + pytest.fail("response.completed was not emitted") + + final_tool_items = [ + event.item + for event in final_events + if event.type + in { + ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + } + and getattr(event.item, "type", None) == "function_call" + ] + assert [item.id for item in final_tool_items] == streamed_ids + assert [item.call_id for item in final_tool_items] == streamed_ids + + argument_event_ids = [ + event.item_id + for event in final_events + if event.type + in { + ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, + ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, + } + ] + assert argument_event_ids + assert set(argument_event_ids) == set(streamed_ids) + + completed = final_events[-1] + completed_calls = [item for item in completed.response.output if item.type == "function_call"] + assert [item.id for item in completed_calls] == streamed_ids + assert [item.call_id for item in completed_calls] == streamed_ids + assert not set(terminal_ids) & { + item_id for item in final_tool_items + completed_calls for item_id in (item.id, item.call_id) + } + + +def test_final_events_preserve_distinct_streamed_item_id_and_call_id(): + from litellm.responses.litellm_completion_transformation.custom_tools import ( + build_tool_call_item_kwargs, + ) + from litellm.responses.litellm_completion_transformation.transformation import ( + LiteLLMCompletionResponsesConfig, + ) + + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + terminal_response = ModelResponse( + id="chatcmpl-terminal", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_terminal", + "type": "function", + "function": {"name": "tool", "arguments": '{"value":1}'}, + } + ], + }, + } + ], + ) + + def distinct_item_id_builder(call_id, *args, **kwargs): + item_kwargs = build_tool_call_item_kwargs(call_id, *args, **kwargs) + if call_id == "call_stream": + item_kwargs["id"] = "fc_stream" + return item_kwargs + + with patch( + "litellm.responses.litellm_completion_transformation.streaming_iterator.build_tool_call_item_kwargs", + side_effect=distinct_item_id_builder, + ): + iterator._queue_tool_call_delta_events( + [ + { + "index": 0, + "id": "call_stream", + "type": "function", + "function": {"name": "tool", "arguments": '{"value":'}, + } + ] + ) + streamed_added = next( + event + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED + ) + assert (streamed_added.item.id, streamed_added.item.call_id) == ( + "fc_stream", + "call_stream", + ) + assert { + event.item_id + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA + } == {"fc_stream"} + iterator._pending_tool_events.clear() + iterator._queue_final_tool_call_done_events(terminal_response) + + original_transform = ( + LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response + ) + + def terminal_response_with_distinct_item_id(*args, **kwargs): + response = original_transform(*args, **kwargs) + terminal_call = next(item for item in response.output if item.type == "function_call") + terminal_call.id = "fc_terminal" + terminal_call.call_id = "call_terminal" + return response + + with patch.object( + LiteLLMCompletionResponsesConfig, + "transform_chat_completion_response_to_responses_api_response", + side_effect=terminal_response_with_distinct_item_id, + ): + completed = iterator._emit_response_completed_event(terminal_response) + + assert completed is not None + argument_events = [ + event + for event in iterator._pending_tool_events + if event.type + in { + ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, + ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, + } + ] + assert argument_events + assert {event.item_id for event in argument_events} == {"fc_stream"} + final_item = next( + event.item + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + ) + assert (final_item.id, final_item.call_id) == ("fc_stream", "call_stream") + completed_call = next(item for item in completed.response.output if item.type == "function_call") + assert (completed_call.id, completed_call.call_id) == ("fc_stream", "call_stream") + + +def test_terminal_only_tool_calls_keep_terminal_identity(): + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + terminal_ids = ["call_terminal_a", "call_terminal_b"] + response = ModelResponse( + id="chatcmpl-terminal-only", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": call_id, + "type": "function", + "function": {"name": f"tool_{index}", "arguments": "{}"}, + } + for index, call_id in enumerate(terminal_ids) + ], + }, + } + ], + ) + + iterator._queue_final_tool_call_done_events(response) + added_items = [ + event.item + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED + ] + + assert [item.id for item in added_items] == terminal_ids + assert [item.call_id for item in added_items] == terminal_ids + + +def test_terminal_only_call_is_not_conflated_with_later_streamed_call(): + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + iterator._queue_tool_call_delta_events( + [ + { + "index": 1, + "id": "call_streamed", + "type": "function", + "function": {"name": "streamed_tool", "arguments": "{}"}, + } + ] + ) + iterator._pending_tool_events.clear() + response = ModelResponse( + id="chatcmpl-mixed", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_terminal_only", + "type": "function", + "function": {"name": "terminal_tool", "arguments": "{}"}, + }, + { + "id": "call_terminal_drifted", + "type": "function", + "function": {"name": "streamed_tool", "arguments": "{}"}, + }, + ], + }, + } + ], + ) + + iterator._queue_final_tool_call_done_events(response) + added_or_done_items = [ + event.item + for event in iterator._pending_tool_events + if event.type + in { + ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + } + ] + completed = iterator._emit_response_completed_event(response) + + assert completed is not None + assert {item.id for item in added_or_done_items} == { + "call_terminal_only", + "call_streamed", + } + completed_calls = [item for item in completed.response.output if item.type == "function_call"] + assert [item.id for item in completed_calls] == [ + "call_terminal_only", + "call_streamed", + ] + assert [item.call_id for item in completed_calls] == [ + "call_terminal_only", + "call_streamed", + ] + + def test_reused_index_with_new_call_id_marks_fallback_ambiguous(): iterator = LiteLLMCompletionStreamingIterator( model="test-model", From 5be81c1e609623821f05cce506216af5c87867b7 Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 17:00:34 -0400 Subject: [PATCH 02/10] refactor(responses): isolate streamed tool call setup --- .../streaming_iterator.py | 63 +++++++++++++------ 1 file changed, 43 insertions(+), 20 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index b315d523f81..631f268386c 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -191,6 +191,43 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return tool_name, namespace return fn_name, None + def _queue_first_streamed_tool_call_event( + self, + call_id: str, + tool_name: str, + tool_namespace: str | None, + output_index: int, + ) -> None: + """Initialize a streamed call and queue its first Responses output item.""" + if call_id in self._tool_args_by_call_id: + return + + self._tool_args_by_call_id[call_id] = "" + streamed_call_ids = getattr(self, "_streamed_tool_call_ids_in_order", None) + if streamed_call_ids is None: + streamed_call_ids = self._streamed_tool_call_ids_in_order = [] + streamed_call_ids.append(call_id) + + item_kwargs = build_tool_call_item_kwargs( + call_id, + tool_name, + "", + "in_progress", + self._custom_tool_names, + ) + self._tool_item_id_by_call_id[call_id] = item_kwargs["id"] + if tool_namespace: + item_kwargs["namespace"] = tool_namespace + + self._sequence_number += 1 + event = OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=output_index, + item=BaseLiteLLMOpenAIResponseObject(**item_kwargs), + ) + event.__dict__["sequence_number"] = self._sequence_number + self._pending_tool_events.append(event) + def _is_reasoning_end(self, chunk): delta: Final = chunk.choices[0].delta @@ -252,26 +289,12 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): tool_name, tool_namespace = self._responses_namespace_tool_call_fields(fn_name) output_index = self._get_or_assign_tool_output_index(call_id) - - if call_id not in self._tool_args_by_call_id: - self._tool_args_by_call_id[call_id] = "" - streamed_call_ids = getattr(self, "_streamed_tool_call_ids_in_order", None) - if streamed_call_ids is None: - streamed_call_ids = self._streamed_tool_call_ids_in_order = [] - streamed_call_ids.append(call_id) - self._sequence_number += 1 - names = self._custom_tool_names - item_kwargs = build_tool_call_item_kwargs(call_id, tool_name, "", "in_progress", names) - self._tool_item_id_by_call_id[call_id] = item_kwargs["id"] - if tool_namespace: - item_kwargs["namespace"] = tool_namespace - event = OutputItemAddedEvent( - type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, - output_index=output_index, - item=BaseLiteLLMOpenAIResponseObject(**item_kwargs), - ) - event.__dict__["sequence_number"] = self._sequence_number - self._pending_tool_events.append(event) + self._queue_first_streamed_tool_call_event( + call_id, + tool_name, + tool_namespace, + output_index, + ) if fn_args_delta: self._tool_args_by_call_id[call_id] += fn_args_delta From bfca46a7dbca6af2fa369b94b6c2eed28f444aad Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 17:11:29 -0400 Subject: [PATCH 03/10] refactor(responses): align streamed identities immutably --- .../streaming_iterator.py | 62 +++++++++++-------- .../test_litellm_completion_responses.py | 3 +- 2 files changed, 37 insertions(+), 28 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 631f268386c..193c4f6417c 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -1,6 +1,7 @@ import time import uuid from collections.abc import Sequence +from itertools import count from typing import Any, Final, cast import litellm @@ -116,8 +117,8 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._tool_args_by_call_id: dict[str, str] = {} self._tool_item_id_by_call_id: dict[str, str] = {} # mutable-ok: filled per call id as tool call events stream self._tool_call_id_by_index: dict[int, str] = {} - self._streamed_tool_call_ids_in_order: list[str] = [] - self._resolved_tool_call_id_by_position: dict[int, str] = {} + self._streamed_tool_call_ids_in_order: list[str] = [] # mutable-ok: accumulates call ids across stream chunks + self._resolved_tool_call_id_by_position: dict[int, str] = {} # mutable-ok: terminal correlation state self._ambiguous_tool_call_indexes: set[int] = set() self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item self._final_tool_events_queued: bool = False @@ -203,12 +204,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return self._tool_args_by_call_id[call_id] = "" - streamed_call_ids = getattr(self, "_streamed_tool_call_ids_in_order", None) - if streamed_call_ids is None: - streamed_call_ids = self._streamed_tool_call_ids_in_order = [] - streamed_call_ids.append(call_id) + self._streamed_tool_call_ids_in_order.append(call_id) - item_kwargs = build_tool_call_item_kwargs( + item_kwargs: Final = build_tool_call_item_kwargs( call_id, tool_name, "", @@ -220,7 +218,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): item_kwargs["namespace"] = tool_namespace self._sequence_number += 1 - event = OutputItemAddedEvent( + event: Final = OutputItemAddedEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, output_index=output_index, item=BaseLiteLLMOpenAIResponseObject(**item_kwargs), @@ -337,10 +335,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if not call_id_raw: continue call_id = self._streamed_tool_call_id_for_terminal_call(tc, position) or str(call_id_raw) - resolved_call_ids = getattr(self, "_resolved_tool_call_id_by_position", None) - if resolved_call_ids is None: - resolved_call_ids = self._resolved_tool_call_id_by_position = {} - resolved_call_ids[position] = call_id + self._resolved_tool_call_id_by_position[position] = call_id output_index = self._get_or_assign_tool_output_index(call_id) fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) @@ -1215,6 +1210,27 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): chat_completion_delta: Final[ChatCompletionDelta] = choice.delta return chat_completion_delta.content or "" + def _output_item_with_streamed_tool_identity(self, item: object, tool_position: int) -> object: + resolved_call_id: Final = self._resolved_tool_call_id_by_position.get(tool_position) + streamed_call_id: Final = ( + resolved_call_id if resolved_call_id is not None else self._streamed_tool_call_id_at_position(tool_position) + ) + if streamed_call_id is None: + return item + + streamed_item_id: Final = self._tool_item_id_by_call_id.get( + streamed_call_id, + getattr(item, "id", streamed_call_id), + ) + identity_update: Final = { # mutable-ok: Pydantic model_copy requires a mapping update payload + "id": streamed_item_id, + "call_id": streamed_call_id, + } + copy_with_identity: Final = getattr(item, "model_copy", None) + if not callable(copy_with_identity): + return item + return copy_with_identity(update=identity_update) + def _output_with_streamed_item_ids(self, responses_api_response: ResponsesAPIResponse) -> tuple[Any, ...]: """ Reuse the item IDs already emitted by the incremental streaming events in the @@ -1231,21 +1247,13 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): "reasoning", self._cached_reasoning_item_id, ) - tool_position = 0 - aligned_items: list[Any] = [] - for item in reasoning_aligned: - if getattr(item, "type", None) in {"function_call", "custom_tool_call"}: - resolved_call_ids: Final = getattr(self, "_resolved_tool_call_id_by_position", {}) - streamed_call_id = resolved_call_ids.get(tool_position) - if streamed_call_id is None: - streamed_call_id = self._streamed_tool_call_id_at_position(tool_position) - tool_position += 1 - if streamed_call_id is not None: - tool_item_ids: Final = getattr(self, "_tool_item_id_by_call_id", {}) - streamed_item_id = tool_item_ids.get(streamed_call_id, getattr(item, "id", streamed_call_id)) - item = item.model_copy(update={"id": streamed_item_id, "call_id": streamed_call_id}) - aligned_items.append(item) - return tuple(aligned_items) + tool_positions: Final = count() + return tuple( + self._output_item_with_streamed_tool_identity(item, next(tool_positions)) + if getattr(item, "type", None) in ("function_call", "custom_tool_call") + else item + for item in reasoning_aligned + ) def _emit_response_completed_event(self, litellm_model_response: ModelResponse) -> ResponseCompletedEvent | None: if litellm_model_response: diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py index b2b8eb5da80..e5d8a0075ca 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py @@ -2,7 +2,6 @@ import json import pytest - from litellm.responses.litellm_completion_transformation.transformation import ( TOOL_CALLS_CACHE, LiteLLMCompletionResponsesConfig, @@ -3411,6 +3410,8 @@ class TestEnsureOutputItemContentPartAdded: iterator._tool_args_by_call_id = {} iterator._tool_item_id_by_call_id = {} iterator._tool_call_id_by_index = {} + iterator._streamed_tool_call_ids_in_order = [] + iterator._resolved_tool_call_id_by_position = {} iterator._ambiguous_tool_call_indexes = set() iterator._next_tool_output_index = 1 iterator._final_tool_events_queued = False From ea167db45ce48ef9691544f4e0cc3959d5d61f02 Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 17:18:55 -0400 Subject: [PATCH 04/10] test(responses): exercise real tool call identity shape --- .../test_streaming_iterator_transformation.py | 97 ++++++------------- 1 file changed, 29 insertions(+), 68 deletions(-) diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 1b4bafdbdd7..25a2cd9a6ab 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -11,7 +11,7 @@ spend tracking stores, so a follow-up previous_response_id still finds the conve """ import json -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock import pytest @@ -510,13 +510,6 @@ def test_final_tool_events_and_completed_snapshot_reuse_streamed_call_identity() def test_final_events_preserve_distinct_streamed_item_id_and_call_id(): - from litellm.responses.litellm_completion_transformation.custom_tools import ( - build_tool_call_item_kwargs, - ) - from litellm.responses.litellm_completion_transformation.transformation import ( - LiteLLMCompletionResponsesConfig, - ) - iterator = LiteLLMCompletionStreamingIterator( model="test-model", litellm_custom_stream_wrapper=AsyncMock(), @@ -546,61 +539,31 @@ def test_final_events_preserve_distinct_streamed_item_id_and_call_id(): } ], ) - - def distinct_item_id_builder(call_id, *args, **kwargs): - item_kwargs = build_tool_call_item_kwargs(call_id, *args, **kwargs) - if call_id == "call_stream": - item_kwargs["id"] = "fc_stream" - return item_kwargs - - with patch( - "litellm.responses.litellm_completion_transformation.streaming_iterator.build_tool_call_item_kwargs", - side_effect=distinct_item_id_builder, - ): - iterator._queue_tool_call_delta_events( - [ - { - "index": 0, - "id": "call_stream", - "type": "function", - "function": {"name": "tool", "arguments": '{"value":'}, - } - ] - ) - streamed_added = next( - event - for event in iterator._pending_tool_events - if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED - ) - assert (streamed_added.item.id, streamed_added.item.call_id) == ( - "fc_stream", - "call_stream", - ) - assert { - event.item_id - for event in iterator._pending_tool_events - if event.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA - } == {"fc_stream"} - iterator._pending_tool_events.clear() - iterator._queue_final_tool_call_done_events(terminal_response) - - original_transform = ( - LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response + iterator._queue_tool_call_delta_events( + [ + { + "index": 0, + "id": "call_stream", + "type": "function", + "function": {"name": "tool", "arguments": '{"value":'}, + } + ] ) - - def terminal_response_with_distinct_item_id(*args, **kwargs): - response = original_transform(*args, **kwargs) - terminal_call = next(item for item in response.output if item.type == "function_call") - terminal_call.id = "fc_terminal" - terminal_call.call_id = "call_terminal" - return response - - with patch.object( - LiteLLMCompletionResponsesConfig, - "transform_chat_completion_response_to_responses_api_response", - side_effect=terminal_response_with_distinct_item_id, - ): - completed = iterator._emit_response_completed_event(terminal_response) + streamed_added = next( + event for event in iterator._pending_tool_events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED + ) + assert (streamed_added.item.id, streamed_added.item.call_id) == ( + "fc_call_stream", + "call_stream", + ) + assert { + event.item_id + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA + } == {"fc_call_stream"} + iterator._pending_tool_events.clear() + iterator._queue_final_tool_call_done_events(terminal_response) + completed = iterator._emit_response_completed_event(terminal_response) assert completed is not None argument_events = [ @@ -613,15 +576,13 @@ def test_final_events_preserve_distinct_streamed_item_id_and_call_id(): } ] assert argument_events - assert {event.item_id for event in argument_events} == {"fc_stream"} + assert {event.item_id for event in argument_events} == {"fc_call_stream"} final_item = next( - event.item - for event in iterator._pending_tool_events - if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + event.item for event in iterator._pending_tool_events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE ) - assert (final_item.id, final_item.call_id) == ("fc_stream", "call_stream") + assert (final_item.id, final_item.call_id) == ("fc_call_stream", "call_stream") completed_call = next(item for item in completed.response.output if item.type == "function_call") - assert (completed_call.id, completed_call.call_id) == ("fc_stream", "call_stream") + assert (completed_call.id, completed_call.call_id) == ("fc_call_stream", "call_stream") def test_terminal_only_tool_calls_keep_terminal_identity(): From 3fb1128a5f6391d339d35a94bd30f8127543909c Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 20:32:05 -0400 Subject: [PATCH 05/10] fix(responses): correlate completed tools by call id --- .../streaming_iterator.py | 19 +++- .../test_litellm_completion_responses.py | 1 + .../test_streaming_iterator_transformation.py | 99 +++++++++++++++++++ 3 files changed, 116 insertions(+), 3 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 193c4f6417c..3e138429f99 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -119,6 +119,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._tool_call_id_by_index: dict[int, str] = {} self._streamed_tool_call_ids_in_order: list[str] = [] # mutable-ok: accumulates call ids across stream chunks self._resolved_tool_call_id_by_position: dict[int, str] = {} # mutable-ok: terminal correlation state + self._streamed_call_id_by_terminal_id: dict[str, str] = {} # mutable-ok: terminal identity correlation self._ambiguous_tool_call_indexes: set[int] = set() self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item self._final_tool_events_queued: bool = False @@ -334,8 +335,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None) if not call_id_raw: continue - call_id = self._streamed_tool_call_id_for_terminal_call(tc, position) or str(call_id_raw) + terminal_call_id = str(call_id_raw) + call_id = self._streamed_tool_call_id_for_terminal_call(tc, position) or terminal_call_id self._resolved_tool_call_id_by_position[position] = call_id + self._streamed_call_id_by_terminal_id[terminal_call_id] = call_id output_index = self._get_or_assign_tool_output_index(call_id) fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) @@ -1211,9 +1214,19 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return chat_completion_delta.content or "" def _output_item_with_streamed_tool_identity(self, item: object, tool_position: int) -> object: - resolved_call_id: Final = self._resolved_tool_call_id_by_position.get(tool_position) + terminal_call_id: Final = getattr(item, "call_id", None) + resolved_by_call_id: Final = ( + self._streamed_call_id_by_terminal_id.get(terminal_call_id) if isinstance(terminal_call_id, str) else None + ) + resolved_by_position: Final = self._resolved_tool_call_id_by_position.get(tool_position) streamed_call_id: Final = ( - resolved_call_id if resolved_call_id is not None else self._streamed_tool_call_id_at_position(tool_position) + resolved_by_call_id + if resolved_by_call_id is not None + else ( + resolved_by_position + if resolved_by_position is not None + else self._streamed_tool_call_id_at_position(tool_position) + ) ) if streamed_call_id is None: return item diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py index e5d8a0075ca..e2569c588df 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py @@ -3412,6 +3412,7 @@ class TestEnsureOutputItemContentPartAdded: iterator._tool_call_id_by_index = {} iterator._streamed_tool_call_ids_in_order = [] iterator._resolved_tool_call_id_by_position = {} + iterator._streamed_call_id_by_terminal_id = {} iterator._ambiguous_tool_call_indexes = set() iterator._next_tool_output_index = 1 iterator._final_tool_events_queued = False diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 25a2cd9a6ab..855612a5eda 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -705,6 +705,105 @@ def test_terminal_only_call_is_not_conflated_with_later_streamed_call(): ] +def test_completed_snapshot_correlates_function_after_server_tool_replacement(): + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + iterator._queue_tool_call_delta_events( + [ + { + "index": 0, + "id": "call_exec_stream", + "type": "function", + "function": { + "name": "bash_code_execution", + "arguments": '{"command":"printf server"}', + }, + }, + { + "index": 1, + "id": "call_regular_stream", + "type": "function", + "function": { + "name": "lookup_weather", + "arguments": '{"city":"Paris"}', + }, + }, + ] + ) + iterator._pending_tool_events.clear() + terminal_response = ModelResponse( + id="chatcmpl-terminal", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "srvtoolu_exec_terminal", + "type": "function", + "function": { + "name": "bash_code_execution", + "arguments": '{"command":"printf server"}', + }, + }, + { + "id": "call_regular_terminal", + "type": "function", + "function": { + "name": "lookup_weather", + "arguments": '{"city":"Paris"}', + }, + }, + ], + "provider_specific_fields": { + "code_interpreter_results": [ + { + "type": "code_interpreter_call", + "id": "srvtoolu_exec_terminal", + "code": "printf server", + "container_id": None, + "status": "completed", + "outputs": [{"type": "logs", "logs": "server"}], + } + ] + }, + }, + } + ], + ) + + iterator._queue_final_tool_call_done_events(terminal_response) + completed = iterator._emit_response_completed_event(terminal_response) + + assert completed is not None + code_calls = [item for item in completed.response.output if item.type == "code_interpreter_call"] + function_calls = [item for item in completed.response.output if item.type == "function_call"] + assert len(code_calls) == 1 + assert len(function_calls) == 1 + code_call = code_calls[0] + assert code_call.id == "srvtoolu_exec_terminal" + assert code_call.code == "printf server" + assert code_call.container_id is None + assert code_call.outputs[0].logs == "server" + function_call = function_calls[0] + assert (function_call.id, function_call.call_id) == ( + "fc_call_regular_stream", + "call_regular_stream", + ) + assert function_call.name == "lookup_weather" + assert function_call.arguments == '{"city":"Paris"}' + + def test_reused_index_with_new_call_id_marks_fallback_ambiguous(): iterator = LiteLLMCompletionStreamingIterator( model="test-model", From 51bc802c1fef1f81ae419979f24bfdcc7dedb19a Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 21:22:50 -0400 Subject: [PATCH 06/10] fix(responses): preserve identity across replacement deltas --- .../streaming_iterator.py | 49 ++++++------- .../test_streaming_iterator_transformation.py | 70 ++++++++++++++----- 2 files changed, 76 insertions(+), 43 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 3e138429f99..d0ec1114511 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -119,8 +119,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._tool_call_id_by_index: dict[int, str] = {} self._streamed_tool_call_ids_in_order: list[str] = [] # mutable-ok: accumulates call ids across stream chunks self._resolved_tool_call_id_by_position: dict[int, str] = {} # mutable-ok: terminal correlation state - self._streamed_call_id_by_terminal_id: dict[str, str] = {} # mutable-ok: terminal identity correlation - self._ambiguous_tool_call_indexes: set[int] = set() + self._streamed_call_id_by_provider_id: dict[str, str] = {} # mutable-ok: provider identity correlation self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item self._final_tool_events_queued: bool = False self._sequence_number: int = 0 @@ -159,8 +158,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return None def _streamed_tool_call_id_at_position(self, position: int) -> str | None: - if position in self._ambiguous_tool_call_indexes: - return None indexed_call_id: Final = self._tool_call_id_by_index.get(position) if indexed_call_id is not None: return indexed_call_id @@ -177,8 +174,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): """Match a terminal aggregate tool call to the identity emitted while streaming.""" tool_call_index: Final = self._normalize_tool_call_index(tool_call) if tool_call_index is not None: - if tool_call_index in self._ambiguous_tool_call_indexes: - return None indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) if indexed_call_id is not None: return indexed_call_id @@ -186,6 +181,23 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return None return self._streamed_tool_call_id_at_position(position) + def _resolve_streamed_tool_call_id(self, tool_call_index: int | None, call_id_raw: object) -> str | None: + if tool_call_index is None: + return str(call_id_raw) if call_id_raw else None + + indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) + if indexed_call_id is not None: + if call_id_raw: + self._streamed_call_id_by_provider_id[str(call_id_raw)] = indexed_call_id + return indexed_call_id + + if not call_id_raw: + return None + + call_id: Final = str(call_id_raw) + self._tool_call_id_by_index[tool_call_index] = call_id + return call_id + def _responses_namespace_tool_call_fields(self, fn_name: str) -> tuple[str, str | None]: mapped: Final = self._namespace_tool_names.get(fn_name) if mapped: @@ -255,25 +267,8 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): for tc in tool_calls: tc_index = self._normalize_tool_call_index(tc) call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None) - call_id = "" - - if call_id_raw: - call_id = str(call_id_raw) - if tc_index is not None: - existing_call_id = self._tool_call_id_by_index.get(tc_index) - if existing_call_id is not None and existing_call_id != call_id: - # Reusing the same index for multiple call_ids is ambiguous for id-less deltas. - # Guard against silent misrouting by disabling index fallback for this index. - self._ambiguous_tool_call_indexes.add(tc_index) - self._tool_call_id_by_index[tc_index] = call_id - elif tc_index is not None: - if tc_index in self._ambiguous_tool_call_indexes: - continue - mapped_call_id = self._tool_call_id_by_index.get(tc_index) - if mapped_call_id: - call_id = mapped_call_id - - if not call_id: + call_id = self._resolve_streamed_tool_call_id(tc_index, call_id_raw) + if call_id is None: continue fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) @@ -338,7 +333,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): terminal_call_id = str(call_id_raw) call_id = self._streamed_tool_call_id_for_terminal_call(tc, position) or terminal_call_id self._resolved_tool_call_id_by_position[position] = call_id - self._streamed_call_id_by_terminal_id[terminal_call_id] = call_id + self._streamed_call_id_by_provider_id[terminal_call_id] = call_id output_index = self._get_or_assign_tool_output_index(call_id) fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) @@ -1216,7 +1211,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def _output_item_with_streamed_tool_identity(self, item: object, tool_position: int) -> object: terminal_call_id: Final = getattr(item, "call_id", None) resolved_by_call_id: Final = ( - self._streamed_call_id_by_terminal_id.get(terminal_call_id) if isinstance(terminal_call_id, str) else None + self._streamed_call_id_by_provider_id.get(terminal_call_id) if isinstance(terminal_call_id, str) else None ) resolved_by_position: Final = self._resolved_tool_call_id_by_position.get(tool_position) streamed_call_id: Final = ( diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 855612a5eda..4ed8b478e1c 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -804,7 +804,7 @@ def test_completed_snapshot_correlates_function_after_server_tool_replacement(): assert function_call.arguments == '{"city":"Paris"}' -def test_reused_index_with_new_call_id_marks_fallback_ambiguous(): +def test_reused_index_with_new_call_id_preserves_first_streamed_identity(): iterator = LiteLLMCompletionStreamingIterator( model="test-model", litellm_custom_stream_wrapper=AsyncMock(), @@ -828,39 +828,77 @@ def test_reused_index_with_new_call_id_marks_fallback_ambiguous(): "index": 0, "id": "call_b", "type": "function", - "function": {"name": "tool_b", "arguments": '{"b":'}, + "function": {"name": "tool_a", "arguments": "1"}, } ] ) - # Ambiguous chunk: index reused and id missing. We should skip fallback rather than misroute. iterator._queue_tool_call_delta_events( [ { "index": 0, "type": "function", - "function": {"arguments": "1}"}, + "function": {"arguments": "}"}, } ] ) + streamed_argument_events = [ + event + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA + ] + + terminal_response = ModelResponse( + id="chatcmpl-terminal", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "index": 0, + "id": "call_b", + "type": "function", + "function": {"name": "tool_a", "arguments": '{"a":1}'}, + } + ], + }, + } + ], + ) + iterator._queue_final_tool_call_done_events(terminal_response) + completed = iterator._emit_response_completed_event(terminal_response) all_events = [] while iterator._pending_tool_events: all_events.append(iterator._pending_tool_events.pop(0)) - delta_events = [ - evt - for evt in all_events - if evt.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA + added_items = [event.item for event in all_events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED] + argument_events = [ + event + for event in all_events + if event.type + in { + ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, + ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, + } ] - arguments_by_call_id = {} - for evt in delta_events: - arguments_by_call_id.setdefault(evt.item_id, "") - arguments_by_call_id[evt.item_id] += evt.delta + done_item = next(event.item for event in all_events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE) - assert arguments_by_call_id["fc_call_a"] == '{"a":' - assert arguments_by_call_id["fc_call_b"] == '{"b":' - assert arguments_by_call_id["fc_call_a"] != '{"a":1}' - assert arguments_by_call_id["fc_call_b"] != '{"b":1}' + assert completed is not None + assert [(item.id, item.call_id) for item in added_items] == [("fc_call_a", "call_a")] + assert {event.item_id for event in streamed_argument_events} == {"fc_call_a"} + assert "".join(event.delta for event in streamed_argument_events) == '{"a":1}' + assert {event.item_id for event in argument_events} == {"fc_call_a"} + assert argument_events[-1].arguments == '{"a":1}' + assert (done_item.id, done_item.call_id) == ("fc_call_a", "call_a") + completed_call = next(item for item in completed.response.output if item.type == "function_call") + assert (completed_call.id, completed_call.call_id) == ("fc_call_a", "call_a") @pytest.mark.asyncio From fa9cb193077179ff57123d3c8931adf8b3083e5e Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 21:35:27 -0400 Subject: [PATCH 07/10] fix(responses): preserve iterator identity state compatibility --- .../streaming_iterator.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index d0ec1114511..f2810ac0abe 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -119,7 +119,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._tool_call_id_by_index: dict[int, str] = {} self._streamed_tool_call_ids_in_order: list[str] = [] # mutable-ok: accumulates call ids across stream chunks self._resolved_tool_call_id_by_position: dict[int, str] = {} # mutable-ok: terminal correlation state - self._streamed_call_id_by_provider_id: dict[str, str] = {} # mutable-ok: provider identity correlation + self._streamed_call_id_by_terminal_id: dict[str, str] = {} # mutable-ok: terminal identity correlation self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item self._final_tool_events_queued: bool = False self._sequence_number: int = 0 @@ -188,7 +188,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) if indexed_call_id is not None: if call_id_raw: - self._streamed_call_id_by_provider_id[str(call_id_raw)] = indexed_call_id + self._streamed_call_id_by_terminal_id[str(call_id_raw)] = indexed_call_id return indexed_call_id if not call_id_raw: @@ -333,7 +333,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): terminal_call_id = str(call_id_raw) call_id = self._streamed_tool_call_id_for_terminal_call(tc, position) or terminal_call_id self._resolved_tool_call_id_by_position[position] = call_id - self._streamed_call_id_by_provider_id[terminal_call_id] = call_id + self._streamed_call_id_by_terminal_id[terminal_call_id] = call_id output_index = self._get_or_assign_tool_output_index(call_id) fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) @@ -1211,7 +1211,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def _output_item_with_streamed_tool_identity(self, item: object, tool_position: int) -> object: terminal_call_id: Final = getattr(item, "call_id", None) resolved_by_call_id: Final = ( - self._streamed_call_id_by_provider_id.get(terminal_call_id) if isinstance(terminal_call_id, str) else None + self._streamed_call_id_by_terminal_id.get(terminal_call_id) if isinstance(terminal_call_id, str) else None ) resolved_by_position: Final = self._resolved_tool_call_id_by_position.get(tool_position) streamed_call_id: Final = ( From cddb23dc5d5144525b97973bd77e2a96faaa0e42 Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 22:12:10 -0400 Subject: [PATCH 08/10] fix(responses): enforce tool metadata identity boundaries --- .../streaming_iterator.py | 113 ++++++++++++-- .../test_litellm_completion_responses.py | 1 + .../test_streaming_iterator_transformation.py | 141 ++++++++++++++++-- 3 files changed, 233 insertions(+), 22 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index f2810ac0abe..5542760a738 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -1,6 +1,7 @@ import time import uuid from collections.abc import Sequence +from dataclasses import dataclass from itertools import count from typing import Any, Final, cast @@ -51,6 +52,39 @@ from litellm.types.utils import ( ) +@dataclass(frozen=True, slots=True) +class _StreamedToolCallMetadata: + call_type: str | None + tool_name: str | None + tool_namespace: str | None + ambiguous: bool = False + + def matches(self, incoming: "_StreamedToolCallMetadata") -> bool: + return all( + existing_value is None or incoming_value is None or existing_value == incoming_value + for existing_value, incoming_value in zip( + (self.call_type, self.tool_name, self.tool_namespace), + (incoming.call_type, incoming.tool_name, incoming.tool_namespace), + ) + ) + + def merged_with(self, incoming: "_StreamedToolCallMetadata") -> "_StreamedToolCallMetadata": + return _StreamedToolCallMetadata( + call_type=self.call_type or incoming.call_type, + tool_name=self.tool_name or incoming.tool_name, + tool_namespace=self.tool_namespace or incoming.tool_namespace, + ambiguous=self.ambiguous, + ) + + def marked_ambiguous(self) -> "_StreamedToolCallMetadata": + return _StreamedToolCallMetadata( + call_type=self.call_type, + tool_name=self.tool_name, + tool_namespace=self.tool_namespace, + ambiguous=True, + ) + + def _index_of_output_item_type(items: Sequence[object], item_type: str) -> int | None: return next( (index for index, item in enumerate(items) if getattr(item, "type", None) == item_type), @@ -117,6 +151,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._tool_args_by_call_id: dict[str, str] = {} self._tool_item_id_by_call_id: dict[str, str] = {} # mutable-ok: filled per call id as tool call events stream self._tool_call_id_by_index: dict[int, str] = {} + self._tool_call_metadata_by_index: dict[ + int, _StreamedToolCallMetadata + ] = {} # mutable-ok: streamed metadata and ambiguity accumulate across chunks self._streamed_tool_call_ids_in_order: list[str] = [] # mutable-ok: accumulates call ids across stream chunks self._resolved_tool_call_id_by_position: dict[int, str] = {} # mutable-ok: terminal correlation state self._streamed_call_id_by_terminal_id: dict[str, str] = {} # mutable-ok: terminal identity correlation @@ -158,6 +195,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return None def _streamed_tool_call_id_at_position(self, position: int) -> str | None: + indexed_metadata: Final = self._tool_call_metadata_by_index.get(position) + if indexed_metadata is not None and indexed_metadata.ambiguous: + return None indexed_call_id: Final = self._tool_call_id_by_index.get(position) if indexed_call_id is not None: return indexed_call_id @@ -172,8 +212,17 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def _streamed_tool_call_id_for_terminal_call(self, tool_call: object, position: int) -> str | None: """Match a terminal aggregate tool call to the identity emitted while streaming.""" + call_id_raw: Final = tool_call.get("id") if isinstance(tool_call, dict) else getattr(tool_call, "id", None) + if call_id_raw: + call_id_match: Final = self._streamed_call_id_by_terminal_id.get(str(call_id_raw)) + if call_id_match is not None: + return call_id_match + tool_call_index: Final = self._normalize_tool_call_index(tool_call) if tool_call_index is not None: + indexed_metadata: Final = self._tool_call_metadata_by_index.get(tool_call_index) + if indexed_metadata is not None and indexed_metadata.ambiguous: + return None indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) if indexed_call_id is not None: return indexed_call_id @@ -181,22 +230,55 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return None return self._streamed_tool_call_id_at_position(position) - def _resolve_streamed_tool_call_id(self, tool_call_index: int | None, call_id_raw: object) -> str | None: + def _resolve_streamed_tool_call_id( + self, + tool_call_index: int | None, + call_id_raw: object, + metadata: _StreamedToolCallMetadata, + ) -> str | None: if tool_call_index is None: - return str(call_id_raw) if call_id_raw else None + if not call_id_raw: + return None + call_id: Final = str(call_id_raw) + self._streamed_call_id_by_terminal_id[call_id] = call_id + return call_id + + incoming_call_id: Final = str(call_id_raw) if call_id_raw else None + known_call_id: Final = ( + self._streamed_call_id_by_terminal_id.get(incoming_call_id) if incoming_call_id is not None else None + ) indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) if indexed_call_id is not None: - if call_id_raw: - self._streamed_call_id_by_terminal_id[str(call_id_raw)] = indexed_call_id - return indexed_call_id + if known_call_id is not None and known_call_id != indexed_call_id: + return known_call_id + indexed_metadata: Final = self._tool_call_metadata_by_index.get( + tool_call_index, + _StreamedToolCallMetadata(None, None, None), + ) + if incoming_call_id is None: + return None if indexed_metadata.ambiguous else indexed_call_id - if not call_id_raw: + if indexed_metadata.matches(metadata): + self._tool_call_metadata_by_index[tool_call_index] = indexed_metadata.merged_with(metadata) + self._streamed_call_id_by_terminal_id[incoming_call_id] = indexed_call_id + return indexed_call_id + + self._tool_call_metadata_by_index[tool_call_index] = indexed_metadata.marked_ambiguous() + if incoming_call_id == indexed_call_id: + return None + self._streamed_call_id_by_terminal_id[incoming_call_id] = incoming_call_id + return incoming_call_id + + if incoming_call_id is None: return None + if known_call_id is not None: + return known_call_id - call_id: Final = str(call_id_raw) - self._tool_call_id_by_index[tool_call_index] = call_id - return call_id + self._tool_call_id_by_index[tool_call_index] = incoming_call_id + self._tool_call_metadata_by_index[tool_call_index] = metadata + self._streamed_call_id_by_terminal_id[incoming_call_id] = incoming_call_id + return incoming_call_id def _responses_namespace_tool_call_fields(self, fn_name: str) -> tuple[str, str | None]: mapped: Final = self._namespace_tool_names.get(fn_name) @@ -267,10 +349,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): for tc in tool_calls: tc_index = self._normalize_tool_call_index(tc) call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None) - call_id = self._resolve_streamed_tool_call_id(tc_index, call_id_raw) - if call_id is None: - continue - fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) fn_name = "" fn_args_delta = "" @@ -281,6 +359,15 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): fn_name = str(getattr(fn, "name", "") or "") fn_args_delta = serialize_tool_call_arguments(getattr(fn, "arguments", "")) tool_name, tool_namespace = self._responses_namespace_tool_call_fields(fn_name) + call_type_raw = tc.get("type") if isinstance(tc, dict) else getattr(tc, "type", None) + metadata = _StreamedToolCallMetadata( + call_type=str(call_type_raw) if call_type_raw else None, + tool_name=tool_name or None, + tool_namespace=tool_namespace, + ) + call_id = self._resolve_streamed_tool_call_id(tc_index, call_id_raw, metadata) + if call_id is None: + continue output_index = self._get_or_assign_tool_output_index(call_id) self._queue_first_streamed_tool_call_event( diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py index e2569c588df..a106c3ffa53 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py @@ -3410,6 +3410,7 @@ class TestEnsureOutputItemContentPartAdded: iterator._tool_args_by_call_id = {} iterator._tool_item_id_by_call_id = {} iterator._tool_call_id_by_index = {} + iterator._tool_call_metadata_by_index = {} iterator._streamed_tool_call_ids_in_order = [] iterator._resolved_tool_call_id_by_position = {} iterator._streamed_call_id_by_terminal_id = {} diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 4ed8b478e1c..227d672d02c 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -804,7 +804,10 @@ def test_completed_snapshot_correlates_function_after_server_tool_replacement(): assert function_call.arguments == '{"city":"Paris"}' -def test_reused_index_with_new_call_id_preserves_first_streamed_identity(): +@pytest.mark.parametrize("include_replacement_metadata", (True, False)) +def test_reused_index_with_new_call_id_preserves_first_streamed_identity( + include_replacement_metadata: bool, +): iterator = LiteLLMCompletionStreamingIterator( model="test-model", litellm_custom_stream_wrapper=AsyncMock(), @@ -822,15 +825,22 @@ def test_reused_index_with_new_call_id_preserves_first_streamed_identity(): } ] ) + replacement_call = ( + { + "index": 0, + "id": "call_b", + "type": "function", + "function": {"name": "tool_a", "arguments": "1"}, + } + if include_replacement_metadata + else { + "index": 0, + "id": "call_b", + "function": {"arguments": "1"}, + } + ) iterator._queue_tool_call_delta_events( - [ - { - "index": 0, - "id": "call_b", - "type": "function", - "function": {"name": "tool_a", "arguments": "1"}, - } - ] + [replacement_call] ) iterator._queue_tool_call_delta_events( [ @@ -901,6 +911,119 @@ def test_reused_index_with_new_call_id_preserves_first_streamed_identity(): assert (completed_call.id, completed_call.call_id) == ("fc_call_a", "call_a") +@pytest.mark.parametrize( + ("replacement_type", "replacement_name"), + (("function", "tool_b"), ("custom", "tool_a")), +) +def test_reused_index_with_changed_tool_metadata_starts_separate_identity( + replacement_type: str, + replacement_name: str, +): + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + + iterator._queue_tool_call_delta_events( + [ + { + "index": 0, + "id": "call_a", + "type": "function", + "function": {"name": "tool_a", "arguments": '{"safe":'}, + } + ] + ) + iterator._queue_tool_call_delta_events( + [ + { + "index": 0, + "id": "call_b", + "type": "function", + "function": {"name": "tool_a", "arguments": ""}, + } + ] + ) + iterator._queue_tool_call_delta_events( + [ + { + "index": 0, + "id": "call_b", + "type": replacement_type, + "function": {"name": replacement_name, "arguments": '{"privileged":'}, + } + ] + ) + iterator._queue_tool_call_delta_events( + [{"index": 0, "id": "call_b", "function": {"arguments": "true}"}}] + ) + iterator._queue_tool_call_delta_events( + [{"index": 0, "function": {"arguments": "ignored"}}] + ) + terminal_response = ModelResponse( + id="chatcmpl-terminal", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "index": 0, + "id": "call_b", + "type": "function", + "function": { + "name": replacement_name, + "arguments": '{"privileged":true}', + }, + } + ], + }, + } + ], + ) + iterator._queue_final_tool_call_done_events(terminal_response) + completed = iterator._emit_response_completed_event(terminal_response) + + added_items = [ + event.item + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED + ] + delta_events = [ + event + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA + ] + done_item = next( + event.item + for event in iterator._pending_tool_events + if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + ) + + assert completed is not None + assert [(item.id, item.call_id) for item in added_items] == [ + ("fc_call_a", "call_a"), + ("fc_call_b", "call_b"), + ] + assert "".join(event.delta for event in delta_events if event.item_id == "fc_call_a") == '{"safe":' + assert "".join(event.delta for event in delta_events if event.item_id == "fc_call_b") == '{"privileged":true}' + assert (done_item.id, done_item.call_id, done_item.name) == ("fc_call_b", "call_b", replacement_name) + completed_call = next(item for item in completed.response.output if item.type == "function_call") + assert (completed_call.id, completed_call.call_id, completed_call.name) == ( + "fc_call_b", + "call_b", + replacement_name, + ) + + @pytest.mark.asyncio async def test_streaming_events_share_the_chat_completion_response_id(): """ From bd795bf5d93a8d2b8562e7da34e97142e516b7d1 Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 23:34:56 -0400 Subject: [PATCH 09/10] fix(responses): reject tool metadata mutation --- .../streaming_iterator.py | 118 +++++++++++++----- .../test_litellm_completion_responses.py | 1 + .../test_streaming_iterator_transformation.py | 88 +++++++------ 3 files changed, 129 insertions(+), 78 deletions(-) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 5542760a738..93d7b15f1bb 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -57,7 +57,6 @@ class _StreamedToolCallMetadata: call_type: str | None tool_name: str | None tool_namespace: str | None - ambiguous: bool = False def matches(self, incoming: "_StreamedToolCallMetadata") -> bool: return all( @@ -73,15 +72,6 @@ class _StreamedToolCallMetadata: call_type=self.call_type or incoming.call_type, tool_name=self.tool_name or incoming.tool_name, tool_namespace=self.tool_namespace or incoming.tool_namespace, - ambiguous=self.ambiguous, - ) - - def marked_ambiguous(self) -> "_StreamedToolCallMetadata": - return _StreamedToolCallMetadata( - call_type=self.call_type, - tool_name=self.tool_name, - tool_namespace=self.tool_namespace, - ambiguous=True, ) @@ -157,6 +147,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._streamed_tool_call_ids_in_order: list[str] = [] # mutable-ok: accumulates call ids across stream chunks self._resolved_tool_call_id_by_position: dict[int, str] = {} # mutable-ok: terminal correlation state self._streamed_call_id_by_terminal_id: dict[str, str] = {} # mutable-ok: terminal identity correlation + self._pending_tool_call_error: litellm.InternalServerError | None = None self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item self._final_tool_events_queued: bool = False self._sequence_number: int = 0 @@ -195,9 +186,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return None def _streamed_tool_call_id_at_position(self, position: int) -> str | None: - indexed_metadata: Final = self._tool_call_metadata_by_index.get(position) - if indexed_metadata is not None and indexed_metadata.ambiguous: - return None indexed_call_id: Final = self._tool_call_id_by_index.get(position) if indexed_call_id is not None: return indexed_call_id @@ -212,17 +200,23 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def _streamed_tool_call_id_for_terminal_call(self, tool_call: object, position: int) -> str | None: """Match a terminal aggregate tool call to the identity emitted while streaming.""" + tool_call_index: Final = self._normalize_tool_call_index(tool_call) + metadata_index: Final = tool_call_index if tool_call_index is not None else position + streamed_metadata: Final = self._tool_call_metadata_by_index.get(metadata_index) + terminal_metadata: Final = self._tool_call_metadata(tool_call) + if streamed_metadata is not None and not streamed_metadata.matches(terminal_metadata): + streamed_call_id: Final = self._tool_call_id_by_index.get(metadata_index) + if streamed_call_id is not None: + self._queue_tool_call_metadata_error(streamed_call_id, streamed_metadata, metadata_index) + return None + call_id_raw: Final = tool_call.get("id") if isinstance(tool_call, dict) else getattr(tool_call, "id", None) if call_id_raw: call_id_match: Final = self._streamed_call_id_by_terminal_id.get(str(call_id_raw)) if call_id_match is not None: return call_id_match - tool_call_index: Final = self._normalize_tool_call_index(tool_call) if tool_call_index is not None: - indexed_metadata: Final = self._tool_call_metadata_by_index.get(tool_call_index) - if indexed_metadata is not None and indexed_metadata.ambiguous: - return None indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) if indexed_call_id is not None: return indexed_call_id @@ -250,25 +244,23 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): indexed_call_id: Final = self._tool_call_id_by_index.get(tool_call_index) if indexed_call_id is not None: - if known_call_id is not None and known_call_id != indexed_call_id: - return known_call_id indexed_metadata: Final = self._tool_call_metadata_by_index.get( tool_call_index, _StreamedToolCallMetadata(None, None, None), ) + if known_call_id is not None and known_call_id != indexed_call_id: + self._queue_tool_call_metadata_error(indexed_call_id, indexed_metadata, tool_call_index) + return None if incoming_call_id is None: - return None if indexed_metadata.ambiguous else indexed_call_id + return indexed_call_id if indexed_metadata.matches(metadata): self._tool_call_metadata_by_index[tool_call_index] = indexed_metadata.merged_with(metadata) self._streamed_call_id_by_terminal_id[incoming_call_id] = indexed_call_id return indexed_call_id - self._tool_call_metadata_by_index[tool_call_index] = indexed_metadata.marked_ambiguous() - if incoming_call_id == indexed_call_id: - return None - self._streamed_call_id_by_terminal_id[incoming_call_id] = incoming_call_id - return incoming_call_id + self._queue_tool_call_metadata_error(indexed_call_id, indexed_metadata, tool_call_index) + return None if incoming_call_id is None: return None @@ -280,6 +272,64 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._streamed_call_id_by_terminal_id[incoming_call_id] = incoming_call_id return incoming_call_id + def _tool_call_metadata(self, tool_call: object) -> _StreamedToolCallMetadata: + function: Final = ( + tool_call.get("function") if isinstance(tool_call, dict) else getattr(tool_call, "function", None) + ) + function_name_raw: Final = ( + function.get("name") if isinstance(function, dict) else getattr(function, "name", None) + ) + tool_name, tool_namespace = self._responses_namespace_tool_call_fields(str(function_name_raw or "")) + call_type_raw: Final = ( + tool_call.get("type") if isinstance(tool_call, dict) else getattr(tool_call, "type", None) + ) + return _StreamedToolCallMetadata( + call_type=str(call_type_raw) if call_type_raw else None, + tool_name=tool_name or None, + tool_namespace=tool_namespace, + ) + + def _queue_tool_call_metadata_error( + self, + call_id: str, + metadata: _StreamedToolCallMetadata, + tool_call_index: int, + ) -> None: + if self._pending_tool_call_error is not None: + return + + output_index: Final = self._get_or_assign_tool_output_index(call_id) + arguments: Final = self._tool_args_by_call_id.get(call_id, "") + item_kwargs: Final = build_tool_call_item_kwargs( + call_id, + metadata.tool_name or "", + arguments, + "incomplete", + self._custom_tool_names, + ) + item_kwargs["id"] = self._tool_item_id_by_call_id.get(call_id, item_kwargs["id"]) + if metadata.tool_namespace: + item_kwargs["namespace"] = metadata.tool_namespace + + self._sequence_number += 1 + self._pending_tool_events.append( + OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=output_index, + sequence_number=self._sequence_number, + item=BaseLiteLLMOpenAIResponseObject(**item_kwargs), + ) + ) + self._pending_tool_call_error = litellm.InternalServerError( + message=f"Provider changed tool metadata at tool call index {tool_call_index}", + llm_provider=self.custom_llm_provider or "", + model=self.model, + ) + + def _raise_pending_tool_call_error(self) -> None: + if self._pending_tool_call_error is not None: + raise self._pending_tool_call_error + def _responses_namespace_tool_call_fields(self, fn_name: str) -> tuple[str, str | None]: mapped: Final = self._namespace_tool_names.get(fn_name) if mapped: @@ -343,7 +393,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): Note: Some providers (like Bedrock) send tool call arguments in one large chunk. We split these into smaller deltas to match OpenAI's token-by-token streaming behavior. """ - if not isinstance(tool_calls, list): + if self._pending_tool_call_error is not None or not isinstance(tool_calls, list): return for tc in tool_calls: @@ -359,14 +409,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): fn_name = str(getattr(fn, "name", "") or "") fn_args_delta = serialize_tool_call_arguments(getattr(fn, "arguments", "")) tool_name, tool_namespace = self._responses_namespace_tool_call_fields(fn_name) - call_type_raw = tc.get("type") if isinstance(tc, dict) else getattr(tc, "type", None) - metadata = _StreamedToolCallMetadata( - call_type=str(call_type_raw) if call_type_raw else None, - tool_name=tool_name or None, - tool_namespace=tool_namespace, - ) + metadata = self._tool_call_metadata(tc) call_id = self._resolve_streamed_tool_call_id(tc_index, call_id_raw, metadata) if call_id is None: + if self._pending_tool_call_error is not None: + return continue output_index = self._get_or_assign_tool_output_index(call_id) @@ -400,7 +447,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): """ Ensure tool calls that were not streamed as deltas still get emitted before response.completed. """ - if self._final_tool_events_queued: + if self._final_tool_events_queued or self._pending_tool_call_error is not None: return self._final_tool_events_queued = True @@ -419,6 +466,8 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): continue terminal_call_id = str(call_id_raw) call_id = self._streamed_tool_call_id_for_terminal_call(tc, position) or terminal_call_id + if self._pending_tool_call_error is not None: + return self._resolved_tool_call_id_by_position[position] = call_id self._streamed_call_id_by_terminal_id[terminal_call_id] = call_id output_index = self._get_or_assign_tool_output_index(call_id) @@ -937,6 +986,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._queue_final_tool_call_done_events(self.litellm_model_response) if self._pending_tool_events: return self._pending_tool_events.pop(0) + self._raise_pending_tool_call_error() done_event: Final = self.return_default_done_events(self.litellm_model_response) if done_event: @@ -1044,6 +1094,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): # Emit any pending tool events before reading a new chunk if self._pending_tool_events: return self._pending_tool_events.pop(0) + self._raise_pending_tool_call_error() try: chunk = self._take_buffered_chunk() @@ -1151,6 +1202,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): # Emit any pending tool events before reading a new chunk if self._pending_tool_events: return self._pending_tool_events.pop(0) + self._raise_pending_tool_call_error() try: buffered_chunk = self._take_buffered_chunk() if buffered_chunk is not None: diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py index a106c3ffa53..3065a030724 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py @@ -3414,6 +3414,7 @@ class TestEnsureOutputItemContentPartAdded: iterator._streamed_tool_call_ids_in_order = [] iterator._resolved_tool_call_id_by_position = {} iterator._streamed_call_id_by_terminal_id = {} + iterator._pending_tool_call_error = None iterator._ambiguous_tool_call_indexes = set() iterator._next_tool_output_index = 1 iterator._final_tool_events_queued = False diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 227d672d02c..77cdbd56f8f 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -15,6 +15,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest +import litellm from litellm.responses.litellm_completion_transformation.streaming_iterator import ( LiteLLMCompletionStreamingIterator, ) @@ -915,16 +916,11 @@ def test_reused_index_with_new_call_id_preserves_first_streamed_identity( ("replacement_type", "replacement_name"), (("function", "tool_b"), ("custom", "tool_a")), ) -def test_reused_index_with_changed_tool_metadata_starts_separate_identity( +def test_reused_index_with_changed_tool_metadata_fails_closed( replacement_type: str, replacement_name: str, ): - iterator = LiteLLMCompletionStreamingIterator( - model="test-model", - litellm_custom_stream_wrapper=AsyncMock(), - request_input="Test input", - responses_api_request={}, - ) + iterator = _build_iterator([]) iterator._queue_tool_call_delta_events( [ @@ -956,11 +952,34 @@ def test_reused_index_with_changed_tool_metadata_starts_separate_identity( } ] ) + events = [] + with pytest.raises(litellm.InternalServerError, match="changed tool metadata at tool call index 0"): + while True: + events.append(next(iterator)) + + added_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED] + done_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE] + delta_events = [event for event in events if event.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA] + + assert [(item.id, item.call_id, item.name) for item in added_items] == [("fc_call_a", "call_a", "tool_a")] + assert [(item.id, item.call_id, item.name, item.status) for item in done_items] == [ + ("fc_call_a", "call_a", "tool_a", "incomplete") + ] + assert "".join(event.delta for event in delta_events) == '{"safe":' + assert all(event.type != ResponsesAPIStreamEvents.RESPONSE_COMPLETED for event in events) + + +def test_terminal_tool_metadata_drift_fails_closed(): + iterator = _build_iterator([]) iterator._queue_tool_call_delta_events( - [{"index": 0, "id": "call_b", "function": {"arguments": "true}"}}] - ) - iterator._queue_tool_call_delta_events( - [{"index": 0, "function": {"arguments": "ignored"}}] + [ + { + "index": 0, + "id": "call_stream", + "type": "function", + "function": {"name": "tool_a", "arguments": '{"safe":true}'}, + } + ] ) terminal_response = ModelResponse( id="chatcmpl-terminal", @@ -977,12 +996,9 @@ def test_reused_index_with_changed_tool_metadata_starts_separate_identity( "tool_calls": [ { "index": 0, - "id": "call_b", + "id": "call_terminal", "type": "function", - "function": { - "name": replacement_name, - "arguments": '{"privileged":true}', - }, + "function": {"name": "tool_b", "arguments": '{"privileged":true}'}, } ], }, @@ -990,38 +1006,20 @@ def test_reused_index_with_changed_tool_metadata_starts_separate_identity( ], ) iterator._queue_final_tool_call_done_events(terminal_response) - completed = iterator._emit_response_completed_event(terminal_response) - added_items = [ - event.item - for event in iterator._pending_tool_events - if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED - ] - delta_events = [ - event - for event in iterator._pending_tool_events - if event.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA - ] - done_item = next( - event.item - for event in iterator._pending_tool_events - if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE - ) + events = [] + with pytest.raises(litellm.InternalServerError, match="changed tool metadata at tool call index 0"): + while True: + events.append(next(iterator)) - assert completed is not None - assert [(item.id, item.call_id) for item in added_items] == [ - ("fc_call_a", "call_a"), - ("fc_call_b", "call_b"), + added_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED] + done_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE] + + assert [(item.id, item.call_id, item.name) for item in added_items] == [("fc_call_stream", "call_stream", "tool_a")] + assert [(item.id, item.call_id, item.name, item.status) for item in done_items] == [ + ("fc_call_stream", "call_stream", "tool_a", "incomplete") ] - assert "".join(event.delta for event in delta_events if event.item_id == "fc_call_a") == '{"safe":' - assert "".join(event.delta for event in delta_events if event.item_id == "fc_call_b") == '{"privileged":true}' - assert (done_item.id, done_item.call_id, done_item.name) == ("fc_call_b", "call_b", replacement_name) - completed_call = next(item for item in completed.response.output if item.type == "function_call") - assert (completed_call.id, completed_call.call_id, completed_call.name) == ( - "fc_call_b", - "call_b", - replacement_name, - ) + assert all(event.type != ResponsesAPIStreamEvents.RESPONSE_COMPLETED for event in events) @pytest.mark.asyncio From afab33eea9e200ecbad2e1f9f667fd950c98773f Mon Sep 17 00:00:00 2001 From: Dan Loftus Date: Tue, 1 Sep 2026 23:53:52 -0400 Subject: [PATCH 10/10] test(responses): satisfy raises lint --- .../test_streaming_iterator_transformation.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 77cdbd56f8f..794d47113f3 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -954,8 +954,7 @@ def test_reused_index_with_changed_tool_metadata_fails_closed( ) events = [] with pytest.raises(litellm.InternalServerError, match="changed tool metadata at tool call index 0"): - while True: - events.append(next(iterator)) + events.extend(iterator) added_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED] done_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE] @@ -1009,8 +1008,7 @@ def test_terminal_tool_metadata_drift_fails_closed(): events = [] with pytest.raises(litellm.InternalServerError, match="changed tool metadata at tool call index 0"): - while True: - events.append(next(iterator)) + events.extend(iterator) added_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED] done_items = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE]