From 5039e636bdea4101696c8e011eb4024ed048ce15 Mon Sep 17 00:00:00 2001 From: cwang-otto Date: Wed, 20 May 2026 08:29:43 +0800 Subject: [PATCH] fix(router): wrap aresponses streaming iterator for mid-stream fallbacks (#28215) Squash-merged by litellm-agent from cwang-otto's PR. --- litellm/router.py | 405 +++++++++++++++++++++++++++++- tests/test_litellm/test_router.py | 372 ++++++++++++++++++++++++++- 2 files changed, 772 insertions(+), 5 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index 420c9b8a816..8c638aa03b5 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -208,6 +208,15 @@ if TYPE_CHECKING: from litellm.router_strategy.quality_router.quality_router import ( QualityRouter, ) + from litellm.responses.streaming_iterator import ( + BaseResponsesAPIStreamingIterator, + ) + from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject + from litellm.types.llms.openai import ( + ResponseAPIUsage, + ResponseInputParam, + ResponsesAPIResponse, + ) Span = Union[_Span, Any] else: @@ -2207,6 +2216,361 @@ class Router: return FallbackStreamWrapper(stream_with_fallbacks()) + @staticmethod + def _extract_partial_responses_usage( + source_iterator: "BaseResponsesAPIStreamingIterator", + ) -> Optional["ResponseAPIUsage"]: + """ + Best-effort: pull partial token usage from a Responses-API streaming + iterator that errored mid-stream, normalized to ResponseAPIUsage so + the caller can combine without crossing token-naming conventions. + + Two sources, in priority order: + 1. The bridge path (LiteLLMCompletionStreamingIterator) accumulates + chat-completion chunks while streaming — feed them through + stream_chunk_builder to recover chat Usage, then translate + (prompt_tokens → input_tokens, completion_tokens → output_tokens). + 2. The native path (ResponsesAPIStreamingIterator) only has a + completed_response object if the stream reached + RESPONSE_COMPLETED before erroring — uncommon mid-stream but + worth checking. Already ResponseAPIUsage-shaped. + + Returns None when no partial usage is recoverable. + """ + from litellm.responses.litellm_completion_transformation.streaming_iterator import ( + LiteLLMCompletionStreamingIterator, + ) + from litellm.types.llms.openai import ( + ResponseAPIUsage, + ResponseCompletedEvent, + ResponseFailedEvent, + ResponseIncompleteEvent, + ) + + # Bridge subclass is the only iterator that accumulates chat-completion + # chunks. isinstance narrows the type so we can read the attribute + # directly instead of getattr-ing on the base class. + if isinstance(source_iterator, LiteLLMCompletionStreamingIterator): + chunks = source_iterator.collected_chat_completion_chunks + if chunks: + try: + from litellm.main import stream_chunk_builder + + built = stream_chunk_builder(chunks=chunks) + if built is not None and built.usage is not None: + chat = built.usage + # getattr-with-default because the test path may + # substitute a SimpleNamespace lacking some fields; + # real Usage instances always have them. + prompt = int(getattr(chat, "prompt_tokens", 0) or 0) + completion = int(getattr(chat, "completion_tokens", 0) or 0) + total = int( + getattr(chat, "total_tokens", prompt + completion) + or (prompt + completion) + ) + return ResponseAPIUsage( + input_tokens=prompt, + output_tokens=completion, + total_tokens=total, + ) + except Exception: + # Builder is best-effort — fall through to native path. + pass + + # Native path: completed_response is set only if RESPONSE_COMPLETED + # arrived before the error (uncommon mid-stream but worth checking). + # Already ResponseAPIUsage-shaped — return as-is. + completed = source_iterator.completed_response + if isinstance( + completed, + (ResponseCompletedEvent, ResponseFailedEvent, ResponseIncompleteEvent), + ): + return completed.response.usage + return None + + @staticmethod + def _combine_responses_fallback_usage( + fallback_item: "BaseLiteLLMOpenAIResponseObject", + partial_usage: "ResponseAPIUsage", + ) -> None: + """ + Merge partial-stream usage with fallback-stream usage on a + Responses-API streaming event. + + Only mutates events that carry a `response` with a `usage` field + (response.completed / response.failed / response.incomplete). Other + events pass through unchanged. + + Both inputs are ResponseAPIUsage-shaped (see + _extract_partial_responses_usage which normalizes the bridge path), + so we can sum input_tokens / output_tokens / total_tokens directly + and produce a clean ResponseAPIUsage — no token-naming split, no + setattr bypass. + """ + from litellm.types.llms.openai import ( + ResponseAPIUsage, + ResponseCompletedEvent, + ResponseFailedEvent, + ResponseIncompleteEvent, + ) + + if not isinstance( + fallback_item, + (ResponseCompletedEvent, ResponseFailedEvent, ResponseIncompleteEvent), + ): + return + response = fallback_item.response + if response.usage is None: + return + + fb = response.usage + response.usage = ResponseAPIUsage( + input_tokens=(partial_usage.input_tokens or 0) + (fb.input_tokens or 0), + output_tokens=(partial_usage.output_tokens or 0) + (fb.output_tokens or 0), + total_tokens=(partial_usage.total_tokens or 0) + (fb.total_tokens or 0), + ) + + @staticmethod + def _build_responses_continuation_input( + input_val: Optional[Union[str, "ResponseInputParam"]], + generated_content: str, + ) -> "ResponseInputParam": + """ + Convert Responses-API input + partial assistant output into a + continuation input that asks the fallback model to pick up where the + prior assistant message stopped. + + Best effort across providers. The chat-completions path uses + Anthropic's `prefix: True` prefill trick on the assistant message; + the Responses-API input schema has no direct equivalent, so we + append an instruction (developer role) plus a prior assistant + message containing the partial output. Providers without prefill + semantics (OpenAI, Vertex) treat this as conversational context + and may regenerate — same trade-off as the chat-completions path + for non-Anthropic fallbacks. + """ + base: List[Dict[str, Any]] + if isinstance(input_val, str): + base = [ + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": input_val}], + } + ] + elif isinstance(input_val, list): + base = list(input_val) + else: + base = [] + continuation: List[Dict[str, Any]] = [ + { + "type": "message", + "role": "developer", + "content": [ + { + "type": "input_text", + "text": ( + "The previous assistant response was interrupted " + "mid-stream. Continue exactly where it stopped — " + "do not repeat any of its content. Your response " + "must read as a seamless continuation." + ), + } + ], + }, + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": generated_content}], + }, + ] + return cast("ResponseInputParam", base + continuation) + + async def _aresponses_streaming_iterator( + self, + response: "BaseResponsesAPIStreamingIterator", + initial_kwargs: Dict[str, Any], + ) -> "BaseResponsesAPIStreamingIterator": + """ + Wrap a Responses-API streaming iterator so MidStreamFallbackError + triggers the Router's fallback chain (parity with + _acompletion_streaming_iterator for the chat-completions path). + + The Responses-API streaming path goes through + _ageneric_api_call_with_fallbacks rather than _acompletion, so the + returned iterator is never wrapped by the chat completions + fallback handler. Without this wrapper, MidStreamFallbackError + raised mid-stream from the underlying CustomStreamWrapper (used by + LiteLLMCompletionStreamingIterator when the Responses API is + served via the completion bridge) propagates unhandled and the + configured cross-provider fallback never fires. + + Full parity with the chat-completions path: + - Pre-first-chunk: retry with the original input unchanged. + - Partial content: inject a developer instruction + prior + assistant message carrying the generated text so the fallback + model continues rather than restarts. + - Usage combining: merge partial-stream usage onto the fallback's + response.completed event so accounting reflects both attempts. + - Stream cleanup: shielded aclose() on both source and fallback + iterators on terminate. + """ + from litellm.exceptions import MidStreamFallbackError + from litellm.responses.streaming_iterator import ( + BaseResponsesAPIStreamingIterator, + ) + + source_iterator = response + + class FallbackResponsesStreamWrapper(BaseResponsesAPIStreamingIterator): + """ + Subclasses BaseResponsesAPIStreamingIterator only for isinstance + compatibility (proxy + interactions code paths check the type). + Bypasses the parent constructor and delegates iteration to an + async generator. + """ + + def __init__(self, async_generator: AsyncGenerator): + import time + + self._async_generator = async_generator + # Mirror every attribute BaseResponsesAPIStreamingIterator.__init__ + # would have set. The wrapper bypasses super().__init__ (it has no + # httpx.Response of its own and no provider config to drive), so + # we copy from source_iterator where applicable and use safe + # defaults elsewhere. This keeps inherited methods (e.g. + # _check_max_streaming_duration, _handle_failure) safe to call. + self.response = source_iterator.response + self.model = source_iterator.model + self.logging_obj = source_iterator.logging_obj + self.finished = False + self.responses_api_provider_config = ( + source_iterator.responses_api_provider_config + ) + self.completed_response = None + self.start_time = source_iterator.start_time + self._failure_handled = False + self._completed_response_cached = False + self._completed_response_logged = False + self._completed_response_cache_hit = None + self._persist_completed_response_before_logging = True + self._stream_created_time = time.time() + self.litellm_metadata = source_iterator.litellm_metadata + self.custom_llm_provider = source_iterator.custom_llm_provider + self.request_data = source_iterator.request_data + self.call_type = source_iterator.call_type + # Preserve hidden params so response headers (model_id, + # api_base, additional_headers) keep flowing. + self._hidden_params = dict(source_iterator._hidden_params or {}) + + def __aiter__(self): + return self + + async def __anext__(self): + return await self._async_generator.__anext__() + + async def aclose(self): + # async generators always expose aclose — no defensive check needed. + await self._async_generator.aclose() + + async def stream_with_fallbacks(): + fallback_response = None + try: + async for item in source_iterator: + yield item + except MidStreamFallbackError as e: + partial_usage = Router._extract_partial_responses_usage(source_iterator) + try: + model_group = cast(str, initial_kwargs.get("model")) + fallbacks: Optional[List] = initial_kwargs.get( + "fallbacks", self.fallbacks + ) + context_window_fallbacks: Optional[List] = initial_kwargs.get( + "context_window_fallbacks", self.context_window_fallbacks + ) + content_policy_fallbacks: Optional[List] = initial_kwargs.get( + "content_policy_fallbacks", self.content_policy_fallbacks + ) + # Re-enter via the per-attempt helper so the fallback chain + # picks deployments through + # _ageneric_api_call_with_fallbacks_helper. + # original_generic_function is preserved by the caller so + # the helper knows what underlying API to invoke per attempt. + initial_kwargs["original_function"] = ( + self._ageneric_api_call_with_fallbacks_helper + ) + if e.is_pre_first_chunk or not e.generated_content: + # No content generated before the error — retry with the + # original input. Adding a continuation prompt would + # waste tokens and confuse the model. + pass + else: + initial_kwargs["input"] = ( + Router._build_responses_continuation_input( + initial_kwargs.get("input"), + e.generated_content, + ) + ) + # The Responses-API path stores observability metadata + # under "litellm_metadata" (not the default "metadata") — + # see _ageneric_api_call_with_fallbacks. Mirroring that + # here ensures model_group, model_group_alias, and trace + # ids land in the same key litellm.aresponses reads from. + self._update_kwargs_before_fallbacks( + model=model_group, + kwargs=initial_kwargs, + metadata_variable_name="litellm_metadata", + ) + fallback_response = ( + await self.async_function_with_fallbacks_common_utils( + e=e, + disable_fallbacks=False, + fallbacks=fallbacks, + context_window_fallbacks=context_window_fallbacks, + content_policy_fallbacks=content_policy_fallbacks, + model_group=model_group, + args=(), + kwargs=initial_kwargs, + ) + ) + + if hasattr(fallback_response, "__aiter__"): + async for fallback_item in fallback_response: # type: ignore + if partial_usage is not None: + Router._combine_responses_fallback_usage( + fallback_item, partial_usage + ) + yield fallback_item + else: + yield fallback_response + except Exception as fallback_error: + verbose_router_logger.error( + f"Responses streaming fallback also failed: {fallback_error}" + ) + raise fallback_error + finally: + with anyio.CancelScope(shield=True): + if hasattr(source_iterator, "aclose"): + try: + await source_iterator.aclose() # type: ignore[func-returns-value] + except BaseException as exc: + verbose_router_logger.debug( + "stream_with_fallbacks(aresponses): error closing source: %s", + exc, + ) + if fallback_response is not None and hasattr( + fallback_response, "aclose" + ): + try: + await fallback_response.aclose() + except BaseException as exc: + verbose_router_logger.debug( + "stream_with_fallbacks(aresponses): error closing fallback: %s", + exc, + ) + + return FallbackResponsesStreamWrapper(stream_with_fallbacks()) + def _completion_streaming_iterator( # noqa: PLR0915 self, model_response: CustomStreamWrapper, @@ -4253,6 +4617,41 @@ class Router: self.fail_calls[model] += 1 raise e + async def _aresponses_with_streaming_fallbacks( + self, original_function: Callable, **kwargs: Any + ) -> Union["ResponsesAPIResponse", "BaseResponsesAPIStreamingIterator"]: + """ + _ageneric_api_call_with_fallbacks for the Responses API, with the + addition of mid-stream fallback handling. + + When stream=True and the underlying call returns a + BaseResponsesAPIStreamingIterator, wrap it with + _aresponses_streaming_iterator so MidStreamFallbackError raised + during iteration triggers the Router's cross-provider fallback chain. + """ + from litellm.responses.streaming_iterator import ( + BaseResponsesAPIStreamingIterator, + ) + + # Snapshot the request kwargs before _ageneric_api_call_with_fallbacks + # mutates them. The original_generic_function is preserved so the + # per-attempt helper knows which underlying API to call on fallback. + fallback_kwargs: Dict[str, Any] = kwargs.copy() + fallback_kwargs["original_generic_function"] = original_function + + response = await self._ageneric_api_call_with_fallbacks( + original_function=original_function, **kwargs + ) + + if kwargs.get("stream") and isinstance( + response, BaseResponsesAPIStreamingIterator + ): + return await self._aresponses_streaming_iterator( + response=response, + initial_kwargs=fallback_kwargs, + ) + return response + def _generic_api_call_with_fallbacks( self, model: str, original_function: Callable, **kwargs ): @@ -5441,9 +5840,13 @@ class Router: custom_llm_provider=custom_llm_provider, **kwargs, ) + elif call_type == "aresponses": + return await self._aresponses_with_streaming_fallbacks( + original_function=original_function, + **kwargs, + ) elif call_type in ( "anthropic_messages", - "aresponses", "_arealtime", "_aresponses_websocket", "acreate_fine_tuning_job", diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index d8be527689e..5e636b86ed6 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -1741,6 +1741,362 @@ async def test_acompletion_streaming_iterator_pre_first_chunk_skips_continuation assert fallback_kwargs["messages"] == messages +# --------------------------------------------------------------------------- +# Shared helpers for the _aresponses_streaming_iterator test suite. +# --------------------------------------------------------------------------- +def _make_responses_iterator( + *, + chunks=(), + error=None, + bridge=False, + model="gpt-4", + hidden_params=None, + chat_chunks=None, +): + """Build a minimal mock Responses-API streaming iterator. + + Bypasses BaseResponsesAPIStreamingIterator.__init__ but mirrors every + attribute production code reads. Yields *chunks*, then raises *error* + (or StopAsyncIteration). Set bridge=True to inherit from + LiteLLMCompletionStreamingIterator so the wrapper's bridge-path + isinstance check (used by usage extraction) matches. + """ + from litellm.responses.litellm_completion_transformation.streaming_iterator import ( + LiteLLMCompletionStreamingIterator, + ) + from litellm.responses.streaming_iterator import ( + BaseResponsesAPIStreamingIterator, + ) + + base = ( + LiteLLMCompletionStreamingIterator + if bridge + else BaseResponsesAPIStreamingIterator + ) + + class _Iter(base): + def __init__(self): + self._chunks = list(chunks) + self._idx = 0 + self._hidden_params = hidden_params or {} + self.model = model + self.custom_llm_provider = "anthropic" + self.logging_obj = MagicMock() + self.litellm_metadata = None + self.responses_api_provider_config = None + self.finished = False + self.completed_response = None + self.response = None + self.start_time = None + self.request_data = {} + self.call_type = None + if chat_chunks is not None: + self.collected_chat_completion_chunks = chat_chunks + + def __aiter__(self): + return self + + async def __anext__(self): + if self._idx < len(self._chunks): + self._idx += 1 + return self._chunks[self._idx - 1] + if error is not None: + raise error + raise StopAsyncIteration + + return _Iter() + + +class _AsyncList: + """Generic async iterator over a list — used as the fallback response.""" + + def __init__(self, items=()): + self._items = list(items) + self._idx = 0 + + def __aiter__(self): + return self + + async def __anext__(self): + if self._idx >= len(self._items): + raise StopAsyncIteration + item = self._items[self._idx] + self._idx += 1 + return item + + +def _make_router_with_fallback(primary="gpt-4", secondary="gpt-3.5-turbo"): + return litellm.Router( + model_list=[ + { + "model_name": primary, + "litellm_params": {"model": primary, "api_key": "k1"}, + }, + { + "model_name": secondary, + "litellm_params": {"model": secondary, "api_key": "k2"}, + }, + ], + fallbacks=[{primary: [secondary]}], + ) + + +@pytest.mark.asyncio +async def test_aresponses_streaming_iterator_fallback(): + """Catches MidStreamFallbackError, re-enters the fallback chain via + async_function_with_fallbacks_common_utils with the per-attempt helper + and original_generic_function preserved. Mirrors + test_acompletion_streaming_iterator for the aresponses path.""" + from litellm.exceptions import MidStreamFallbackError + from litellm.responses.streaming_iterator import ( + BaseResponsesAPIStreamingIterator, + ) + + router = _make_router_with_fallback( + "anthropic/claude-sonnet-4-6", "vertex_ai/claude-sonnet-4-6" + ) + src = _make_responses_iterator( + chunks=[MagicMock(type="response.created")], + error=MidStreamFallbackError( + message="anthropic socket timeout", + model="anthropic/claude-sonnet-4-6", + llm_provider="anthropic", + is_pre_first_chunk=False, + generated_content="", + ), + model="anthropic/claude-sonnet-4-6", + hidden_params={"model_id": "src-deployment-1"}, + ) + fallback_chunks = [ + MagicMock(type="response.output_text.delta"), + MagicMock(type="response.completed"), + ] + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + return_value=_AsyncList(fallback_chunks), + ) as mock_fallback_utils: + wrapped = await router._aresponses_streaming_iterator( + response=src, + initial_kwargs={ + "model": "anthropic/claude-sonnet-4-6", + "stream": True, + "input": "Hi", + "original_generic_function": litellm.aresponses, + }, + ) + assert isinstance(wrapped, BaseResponsesAPIStreamingIterator) + assert wrapped._hidden_params.get("model_id") == "src-deployment-1" + collected = [c async for c in wrapped] + + assert len(collected) == 3 # 1 primary chunk + 2 fallback chunks + call_kwargs = mock_fallback_utils.call_args.kwargs + fbk = call_kwargs["kwargs"] + # Bound methods compare equal when they share the same instance + __func__. + assert fbk["original_function"] == router._ageneric_api_call_with_fallbacks_helper + assert fbk["original_generic_function"] is litellm.aresponses + assert call_kwargs["model_group"] == "anthropic/claude-sonnet-4-6" + assert call_kwargs["disable_fallbacks"] is False + + +@pytest.mark.asyncio +async def test_aresponses_streaming_iterator_writes_litellm_metadata_on_fallback(): + """Regression: model_group must land under "litellm_metadata" (the key + litellm.aresponses reads), not the default "metadata".""" + from litellm.exceptions import MidStreamFallbackError + + router = _make_router_with_fallback() + src = _make_responses_iterator( + error=MidStreamFallbackError( + message="boom", + model="gpt-4", + llm_provider="anthropic", + is_pre_first_chunk=True, + generated_content="", + ) + ) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + return_value=_AsyncList(), + ) as mock_fallback_utils: + wrapped = await router._aresponses_streaming_iterator( + response=src, + initial_kwargs={ + "model": "gpt-4", + "stream": True, + "input": "Hello", + "original_generic_function": litellm.aresponses, + }, + ) + async for _ in wrapped: + pass + + fbk = mock_fallback_utils.call_args.kwargs["kwargs"] + assert "litellm_metadata" in fbk, "wrong metadata_variable_name" + assert fbk["litellm_metadata"]["model_group"] == "gpt-4" + assert "model_group" not in fbk.get( + "metadata", {} + ), "model_group leaked into 'metadata' instead of 'litellm_metadata'" + + +@pytest.mark.asyncio +async def test_aresponses_streaming_iterator_pre_first_chunk_skips_continuation(): + """Pre-first-chunk error: original input is preserved unchanged.""" + from litellm.exceptions import MidStreamFallbackError + + router = _make_router_with_fallback() + src = _make_responses_iterator( + error=MidStreamFallbackError( + message="socket timeout before first chunk", + model="gpt-4", + llm_provider="anthropic", + is_pre_first_chunk=True, + generated_content="", + ) + ) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + return_value=_AsyncList(), + ) as mock_fallback_utils: + wrapped = await router._aresponses_streaming_iterator( + response=src, + initial_kwargs={ + "model": "gpt-4", + "stream": True, + "input": "Hello", + "original_generic_function": litellm.aresponses, + }, + ) + async for _ in wrapped: + pass + + fbk = mock_fallback_utils.call_args.kwargs["kwargs"] + assert fbk["input"] == "Hello" # original input, no continuation messages + + +@pytest.mark.asyncio +async def test_aresponses_streaming_iterator_partial_content_injects_continuation(): + """Mid-stream error: input is rewritten to include user prompt + + developer instruction + prior assistant message with partial output.""" + from litellm.exceptions import MidStreamFallbackError + + router = _make_router_with_fallback() + src = _make_responses_iterator( + chunks=[MagicMock(type="response.output_text.delta")], + error=MidStreamFallbackError( + message="socket reset mid-stream", + model="gpt-4", + llm_provider="anthropic", + is_pre_first_chunk=False, + generated_content="The capital of France is", + ), + ) + + with patch.object( + router, + "async_function_with_fallbacks_common_utils", + return_value=_AsyncList(), + ) as mock_fallback_utils: + wrapped = await router._aresponses_streaming_iterator( + response=src, + initial_kwargs={ + "model": "gpt-4", + "stream": True, + "input": "What's the capital of France?", + "original_generic_function": litellm.aresponses, + }, + ) + async for _ in wrapped: + pass + + new_input = mock_fallback_utils.call_args.kwargs["kwargs"]["input"] + assert isinstance(new_input, list) + assert new_input[0]["role"] == "user" + assert new_input[0]["content"][0]["text"] == "What's the capital of France?" + assert new_input[1]["role"] == "developer" + assert "do not repeat" in new_input[1]["content"][0]["text"].lower() + assert new_input[2]["role"] == "assistant" + assert new_input[2]["content"][0]["type"] == "output_text" + assert new_input[2]["content"][0]["text"] == "The capital of France is" + + +@pytest.mark.asyncio +async def test_aresponses_streaming_iterator_combines_partial_usage(): + """Partial usage from the bridge path is normalized to ResponseAPIUsage + and summed onto the fallback's response.completed event — no token-name + split, clean ResponseAPIUsage on output.""" + from types import SimpleNamespace + + from litellm.exceptions import MidStreamFallbackError + from litellm.types.llms.openai import ( + ResponseAPIUsage, + ResponseCompletedEvent, + ResponsesAPIResponse, + ResponsesAPIStreamEvents, + ) + + router = _make_router_with_fallback() + src = _make_responses_iterator( + bridge=True, + chat_chunks=[MagicMock()], + chunks=[MagicMock(type="response.output_text.delta")], + error=MidStreamFallbackError( + message="boom", + model="gpt-4", + llm_provider="anthropic", + is_pre_first_chunk=False, + generated_content="hello", + ), + ) + + fallback_response_object = ResponsesAPIResponse( + id="resp_test", created_at=0, model="gpt-4", object="response", output=[] + ) + fallback_response_object.usage = ResponseAPIUsage( + input_tokens=20, output_tokens=15, total_tokens=35 + ) + fallback_event = ResponseCompletedEvent( + type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, + response=fallback_response_object, + ) + + with ( + patch( + "litellm.main.stream_chunk_builder", + return_value=SimpleNamespace( + usage=SimpleNamespace(prompt_tokens=10, completion_tokens=4) + ), + ), + patch.object( + router, + "async_function_with_fallbacks_common_utils", + return_value=_AsyncList([fallback_event]), + ), + ): + wrapped = await router._aresponses_streaming_iterator( + response=src, + initial_kwargs={ + "model": "gpt-4", + "stream": True, + "input": "hi", + "original_generic_function": litellm.aresponses, + }, + ) + async for _ in wrapped: + pass + + merged = fallback_response_object.usage + assert isinstance(merged, ResponseAPIUsage) + assert merged.input_tokens == 30 # 10 (translated from prompt_tokens) + 20 + assert merged.output_tokens == 19 # 4 (translated from completion_tokens) + 15 + assert merged.total_tokens == 49 + + @pytest.mark.asyncio async def test_async_function_with_fallbacks_common_utils(): """Test the async_function_with_fallbacks_common_utils method""" @@ -3863,7 +4219,15 @@ def test_is_deployment_blocked_static_helper_reflects_blocked_flag(): # No model_info on deployment object → treated as not blocked assert litellm.Router._is_deployment_blocked(object()) is False missing_blocked = types.SimpleNamespace() - assert litellm.Router._is_deployment_blocked(types.SimpleNamespace(model_info=missing_blocked)) is False - assert litellm.Router._is_deployment_blocked( - types.SimpleNamespace(model_info=types.SimpleNamespace(blocked=True)) - ) is True + assert ( + litellm.Router._is_deployment_blocked( + types.SimpleNamespace(model_info=missing_blocked) + ) + is False + ) + assert ( + litellm.Router._is_deployment_blocked( + types.SimpleNamespace(model_info=types.SimpleNamespace(blocked=True)) + ) + is True + )