scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-27 01:22:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 80

fix(router): hold Responses lifecycle events until output so a pre-output fallback announces one response (#43238)

Browse source 
* fix(router): hold Responses lifecycle events until output so a pre-output fallback announces one response

* fix(router): narrow the responses wrapper close guards to Exception and test the hold helpers directly

* test(router): type the responses fallback test helpers

* fix(router): replay the held lifecycle events when the fallback stream fails before its first event

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-25 18:16:05 -07:00 • committed by GitHub
parent e9491d31b5
commit 4adbc13d79
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 338 additions and 88 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/responses/streaming_iterator.py
View file

@ -265,7 +265,7 @@ def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool:
return not isinstance(status_code, int) or status_code >= 500 or status_code == 429
_PRE_OUTPUT_LIFECYCLE_EVENT_TYPES: Final = frozenset({"response.created", "response.in_progress", "response.queued"})
PRE_OUTPUT_LIFECYCLE_EVENT_TYPES: Final = frozenset({"response.created", "response.in_progress", "response.queued"})
class BaseResponsesAPIStreamingIterator:
@ -885,7 +885,7 @@ class BaseResponsesAPIStreamingIterator:
def _note_yielded_event(self, event: ResponsesAPIStreamingResponse) -> None:
self._yielded_first_chunk = True
if event.type not in _PRE_OUTPUT_LIFECYCLE_EVENT_TYPES:
if event.type not in PRE_OUTPUT_LIFECYCLE_EVENT_TYPES:
self._output_started = True
def _fallback_error(self, original: Exception) -> MidStreamFallbackError:

203
litellm/router.py
View file

@ -614,6 +614,17 @@ def _anthropic_stream_commits_now(chunk: object, has_generated_content: bool, bu
return is_anthropic_content_delta_chunk(chunk) or buffered_chunk_count >= MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS
MAX_HELD_PRE_OUTPUT_RESPONSES_EVENTS: Final = 200
def _responses_stream_holds_event(item: object, held_event_count: int) -> bool:
from litellm.responses.streaming_iterator import PRE_OUTPUT_LIFECYCLE_EVENT_TYPES
if held_event_count >= MAX_HELD_PRE_OUTPUT_RESPONSES_EVENTS:
return False
return getattr(item, "type", None) in PRE_OUTPUT_LIFECYCLE_EVENT_TYPES
class FallbackAwareAnthropicMessagesStream:
"""
Bare async generators can't carry the `_hidden_params` attribute the
@ -3332,100 +3343,140 @@ class Router:
await self._async_generator.aclose()
async def stream_with_fallbacks():
fallback_response = None
held_lifecycle_events: tuple[object, ...] = () # rebind-ok: flushed at first output, dropped on fallback
try:
async for item in source_iterator:
if _responses_stream_holds_event(item, len(held_lifecycle_events)):
held_lifecycle_events = (*held_lifecycle_events, item)
continue
for held_event in held_lifecycle_events:
yield held_event
held_lifecycle_events = ()
yield item
for held_event in held_lifecycle_events:
yield held_event
except MidStreamFallbackError as e:
partial_usage: Final = Router._extract_partial_responses_usage(source_iterator)
try:
model_group: Final = cast(str, initial_kwargs.get("model"))
fallbacks: Final[list | None] = initial_kwargs.get("fallbacks", self.fallbacks)
context_window_fallbacks: Final[list | None] = initial_kwargs.get(
"context_window_fallbacks", self.context_window_fallbacks
async with contextlib.aclosing(
self._aresponses_fallback_attempt(
e, source_iterator, initial_kwargs, wrapper.adopt_fallback_headers, held_lifecycle_events
)
content_policy_fallbacks: Final[list | None] = initial_kwargs.get(
"content_policy_fallbacks", self.content_policy_fallbacks
)
initial_kwargs["original_function"] = self._ageneric_api_call_with_fallbacks_responses_attempt
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",
)
# The content-policy dispatch branch matches on the trigger's own type, so a refusal's
# MidStreamFallbackError envelope is unwrapped here or the wrong fallback list is consulted.
fallback_trigger: Final[Exception] = (
e.original_exception
if isinstance(e.original_exception, litellm.ContentPolicyViolationError)
else e
)
fallback_response = await self.async_function_with_fallbacks_common_utils(
e=fallback_trigger,
disable_fallbacks=fallbacks_disabled_for_request(initial_kwargs),
fallbacks=fallbacks,
context_window_fallbacks=context_window_fallbacks,
content_policy_fallbacks=content_policy_fallbacks,
model_group=model_group,
args=(),
kwargs=initial_kwargs,
include_fallback_errors=initial_kwargs.get("include_fallback_errors", False) is True,
)
prepared_fallback_hidden_params = wrapper.adopt_fallback_headers(fallback_response)
if hasattr(fallback_response, "__aiter__"):
async for fallback_item in fallback_response:
Router._apply_fallback_hidden_params_to_item(fallback_item, prepared_fallback_hidden_params)
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("Responses streaming fallback also failed: %s", fallback_error)
if (
isinstance(fallback_error, MidStreamFallbackError)
and fallback_error.original_exception is not None
):
raise fallback_error.original_exception from fallback_error
raise fallback_error
) as fallback_stream:
async for fallback_item in fallback_stream:
yield fallback_item
except Exception:
for held_event in held_lifecycle_events:
yield held_event
raise
finally:
with anyio.CancelScope(shield=True):
if hasattr(source_iterator, "aclose"):
try:
await source_iterator.aclose()
except BaseException as exc:
except Exception 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,
)
wrapper: Final = FallbackResponsesStreamWrapper(stream_with_fallbacks())
return wrapper
async def _aresponses_fallback_attempt(
self,
e: "MidStreamFallbackError",
source_iterator: "BaseResponsesAPIStreamingIterator",
initial_kwargs: dict[str, Any], # mutable-ok: mutated in-place before re-entering the fallback chain
adopt_headers: Callable[[object], tuple[dict[str, object], dict[str, object]]], # mutable-ok: hidden params
held_lifecycle_events: tuple[object, ...],
) -> AsyncGenerator[object, None]:
"""
Re-enters the Router's fallback chain for a mid-stream Responses API error and yields
whatever the fallback attempt produces. The lifecycle events the primary stream held
back reach the client only when no fallback lands, so the client sees exactly one
response announced, the one whose id completes. Split out of
_aresponses_streaming_iterator to keep each function's cyclomatic complexity within
the repo's C901 budget.
"""
from litellm.exceptions import MidStreamFallbackError
partial_usage: Final = Router._extract_partial_responses_usage(source_iterator)
fallback_response = None # rebind-ok: pre-init so finally can close it if a fallback was actually attempted
fallback_yielded = False # rebind-ok: flipped on the first fallback item so a fallback that dies before its first event still replays the primary's held announcement
try:
model_group: Final = cast(str, initial_kwargs.get("model")) # cast-ok: model group
fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the common_utils list|None param
"fallbacks", self.fallbacks
)
context_window_fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the param below
"context_window_fallbacks", self.context_window_fallbacks
)
content_policy_fallbacks: Final[list | None] = initial_kwargs.get( # mutable-ok: matches the param below
"content_policy_fallbacks", self.content_policy_fallbacks
)
initial_kwargs["original_function"] = ( # rebind-ok: the fallback chain re-enters on the same kwargs
self._ageneric_api_call_with_fallbacks_responses_attempt
)
if e.generated_content and not e.is_pre_first_chunk:
initial_kwargs["input"] = Router._build_responses_continuation_input( # rebind-ok: fallback hop 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",
)
# The content-policy dispatch branch matches on the trigger's own type, so a refusal's
# MidStreamFallbackError envelope is unwrapped here or the wrong fallback list is consulted.
fallback_trigger: Final[Exception] = (
e.original_exception if isinstance(e.original_exception, litellm.ContentPolicyViolationError) else e
)
fallback_response = await self.async_function_with_fallbacks_common_utils( # rebind-ok: set on success
e=fallback_trigger,
disable_fallbacks=fallbacks_disabled_for_request(initial_kwargs),
fallbacks=fallbacks,
context_window_fallbacks=context_window_fallbacks,
content_policy_fallbacks=content_policy_fallbacks,
model_group=model_group,
args=(),
kwargs=initial_kwargs,
include_fallback_errors=initial_kwargs.get("include_fallback_errors", False) is True,
)
prepared_fallback_hidden_params: Final = adopt_headers(fallback_response)
if hasattr(fallback_response, "__aiter__"):
async for fallback_item in fallback_response:
Router._apply_fallback_hidden_params_to_item(fallback_item, prepared_fallback_hidden_params)
if partial_usage is not None:
Router._combine_responses_fallback_usage(fallback_item, partial_usage)
fallback_yielded = True
yield fallback_item
else:
fallback_yielded = True # rebind-ok: see the pre-init above
yield fallback_response
except Exception as fallback_error:
verbose_router_logger.error("Responses streaming fallback also failed: %s", fallback_error)
if not fallback_yielded:
for held_event in held_lifecycle_events:
yield held_event
if isinstance(fallback_error, MidStreamFallbackError) and fallback_error.original_exception is not None:
raise fallback_error.original_exception from fallback_error
raise
finally:
if fallback_response is not None and hasattr(fallback_response, "aclose"):
with anyio.CancelScope(shield=True):
try:
await fallback_response.aclose()
except Exception as exc:
verbose_router_logger.debug(
"stream_with_fallbacks(aresponses): error closing fallback: %s",
exc,
)
def _completion_streaming_iterator(
self,
model_response: CustomStreamWrapper,

219
tests/unit/test_router/test_router.py
View file

@ -8,7 +8,7 @@ import os
import sys
import threading
import warnings
from collections.abc import Awaitable, Callable, Mapping
from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Mapping
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from typing import Final, Literal
@ -37,6 +37,7 @@ from litellm.models.access_group import LiteLLM_AccessGroupTable
from litellm.proxy._types import LiteLLM_TeamTable, LitellmUserRoles, Member, ProxyException, UserAPIKeyAuth
from litellm.router import (
MAX_BUFFERED_PRE_CONTENT_ANTHROPIC_CHUNKS,
MAX_HELD_PRE_OUTPUT_RESPONSES_EVENTS,
FallbackAwareAnthropicMessagesStream,
_anthropic_stream_commits_now,
_anthropic_stream_error_is_gateway_verdict,
@ -46,6 +47,7 @@ from litellm.router import (
_anthropic_stream_should_decline_fallback,
_anthropic_stream_should_drop_pre_content_ping,
_is_retriable_anthropic_status,
_responses_stream_holds_event,
)
from litellm.router_strategy import simple_shuffle
from litellm.router_utils.client_initalization_utils import MaxParallelRequestsLimit
@ -4213,6 +4215,7 @@ async def test_aresponses_streaming_iterator_fallback():
hidden_params={"model_id": "src-deployment-1"},
)
fallback_chunks = [
MagicMock(type="response.created"),
MagicMock(type="response.output_text.delta"),
MagicMock(type="response.completed"),
]
@ -4235,7 +4238,7 @@ async def test_aresponses_streaming_iterator_fallback():
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
assert collected == 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__.
@ -4522,20 +4525,35 @@ def _make_native_responses_iterator(*, sse_payloads: tuple[dict[str, str], ...],
_RESPONSES_LIFECYCLE_PAYLOADS: Final = ({"type": "response.created"}, {"type": "response.in_progress"})
async def _events_until_error(stream: AsyncIterable[object]) -> AsyncIterator[object]:
try:
async for chunk in stream:
yield chunk
except Exception as error:
yield error
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_falls_back_on_transport_drop_before_output():
"""A connection lost after response.created but before any output item is re-routed to the
fallback with the original input, the same as a provider error event would be."""
fallback with the original input, the same as a provider error event would be, and the client
sees one response lifecycle: the fallback's, whose id the completed event carries."""
router: Final = _make_router_with_fallback()
src: Final = _make_native_responses_iterator(
sse_payloads=_RESPONSES_LIFECYCLE_PAYLOADS,
trailing_error=httpx.ReadError("Response payload is not completed"),
)
fallback_chunks: Final = [
MagicMock(type="response.created", response=MagicMock(id="resp_fallback")),
MagicMock(type="response.in_progress", response=MagicMock(id="resp_fallback")),
MagicMock(type="response.output_text.delta"),
MagicMock(type="response.completed", response=MagicMock(id="resp_fallback")),
]
with patch.object(
router,
"async_function_with_fallbacks_common_utils",
return_value=_AsyncList([MagicMock(type="response.completed")]),
return_value=_AsyncList(fallback_chunks),
) as mock_fallback_utils:
wrapped: Final = await router._aresponses_streaming_iterator(
response=src,
@ -4546,9 +4564,10 @@ async def test_aresponses_streaming_iterator_falls_back_on_transport_drop_before
"original_generic_function": litellm.aresponses,
},
)
seen: Final = [chunk.type async for chunk in wrapped]
collected: Final = [chunk async for chunk in wrapped]
assert seen == ["response.created", "response.in_progress", "response.completed"]
assert collected == fallback_chunks
assert [chunk.response.id for chunk in collected if chunk.type == "response.created"] == ["resp_fallback"]
assert isinstance(mock_fallback_utils.call_args.kwargs["e"], MidStreamFallbackError)
assert mock_fallback_utils.call_args.kwargs["kwargs"]["input"] == "Hello"
@ -4576,17 +4595,197 @@ async def test_aresponses_streaming_iterator_surfaces_transport_drop_when_no_fal
"original_generic_function": litellm.aresponses,
},
)
with pytest.raises(httpx.ReadError) as exc_info:
async for _ in wrapped:
pass
outcome: Final = [item async for item in _events_until_error(wrapped)]
assert exc_info.value is transport_error
assert [item.type for item in outcome[:-1]] == ["response.created", "response.in_progress"]
assert outcome[-1] is transport_error
assert mock_fallback_utils.await_count == 1
trigger: Final = mock_fallback_utils.await_args.kwargs["e"]
assert isinstance(trigger, MidStreamFallbackError)
assert trigger.original_exception is transport_error
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_forwards_lifecycle_events_in_order_once_output_starts():
router: Final = _make_router_with_fallback()
chunks: Final = [
MagicMock(type="response.created"),
MagicMock(type="response.in_progress"),
MagicMock(type="response.output_text.delta"),
MagicMock(type="response.completed"),
]
wrapped: Final = await router._aresponses_streaming_iterator(
response=_make_responses_iterator(chunks=chunks),
initial_kwargs={"model": "gpt-4", "stream": True, "input": "Hello"},
)
assert [chunk async for chunk in wrapped] == chunks
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_flushes_held_lifecycle_events_when_the_stream_ends_without_output():
router: Final = _make_router_with_fallback()
chunks: Final = [MagicMock(type="response.created"), MagicMock(type="response.in_progress")]
wrapped: Final = await router._aresponses_streaming_iterator(
response=_make_responses_iterator(chunks=chunks),
initial_kwargs={"model": "gpt-4", "stream": True, "input": "Hello"},
)
assert [chunk async for chunk in wrapped] == chunks
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_forwards_held_lifecycle_events_before_a_non_fallback_error():
router: Final = _make_router_with_fallback()
chunks: Final = [MagicMock(type="response.created"), MagicMock(type="response.in_progress")]
client_error: Final = litellm.BadRequestError(message="bad input", model="gpt-4", llm_provider="openai")
wrapped: Final = await router._aresponses_streaming_iterator(
response=_make_responses_iterator(chunks=chunks, error=client_error),
initial_kwargs={"model": "gpt-4", "stream": True, "input": "Hello"},
)
outcome: Final = [item async for item in _events_until_error(wrapped)]
assert outcome[:-1] == chunks
assert outcome[-1] is client_error
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_commits_held_lifecycle_events_at_the_hold_cap():
router: Final = _make_router_with_fallback()
chunks: Final = [MagicMock(type="response.in_progress") for _ in range(MAX_HELD_PRE_OUTPUT_RESPONSES_EVENTS + 1)]
src: Final = _make_responses_iterator(
chunks=chunks,
error=MidStreamFallbackError(
message="dropped before output", model="gpt-4", llm_provider="openai", is_pre_first_chunk=True
),
)
fallback_chunks: Final = [MagicMock(type="response.created"), MagicMock(type="response.completed")]
with patch.object(router, "async_function_with_fallbacks_common_utils", return_value=_AsyncList(fallback_chunks)):
wrapped: Final = await router._aresponses_streaming_iterator(
response=src,
initial_kwargs={"model": "gpt-4", "stream": True, "input": "Hello"},
)
collected: Final = [chunk async for chunk in wrapped]
assert collected == [*chunks, *fallback_chunks]
@pytest.mark.parametrize(
("event_type", "held_event_count", "expected"),
[
("response.created", 0, True),
("response.in_progress", 1, True),
("response.queued", MAX_HELD_PRE_OUTPUT_RESPONSES_EVENTS - 1, True),
("response.in_progress", MAX_HELD_PRE_OUTPUT_RESPONSES_EVENTS, False),
("response.output_item.added", 0, False),
("response.output_text.delta", 0, False),
("response.completed", 0, False),
],
)
def test_responses_stream_holds_event_holds_only_pre_output_lifecycle_events_under_the_cap(
event_type: str, held_event_count: int, expected: bool
):
assert _responses_stream_holds_event(MagicMock(type=event_type), held_event_count) is expected
@pytest.mark.asyncio
async def test_aresponses_fallback_attempt_drops_held_lifecycle_events_when_a_fallback_lands():
router: Final = _make_router_with_fallback()
trigger: Final = MidStreamFallbackError(
message="dropped before output", model="gpt-4", llm_provider="openai", is_pre_first_chunk=True
)
held: Final = (MagicMock(type="response.created"), MagicMock(type="response.in_progress"))
fallback_chunks: Final = [MagicMock(type="response.created"), MagicMock(type="response.completed")]
adopt_headers: Final = MagicMock(return_value=({}, {}))
with patch.object(
router, "async_function_with_fallbacks_common_utils", return_value=_AsyncList(fallback_chunks)
) as mock_fallback_utils:
collected: Final = [
chunk
async for chunk in router._aresponses_fallback_attempt(
trigger,
_make_responses_iterator(),
{"model": "gpt-4", "stream": True, "input": "Hello"},
adopt_headers,
held,
)
]
assert collected == fallback_chunks
adopt_headers.assert_called_once()
assert mock_fallback_utils.await_args.kwargs["e"] is trigger
@pytest.mark.asyncio
async def test_aresponses_fallback_attempt_replays_held_lifecycle_events_when_the_fallback_dies_before_its_first_event():
"""A fallback stream that raises before yielding anything announced no response of its own, so the
primary's held created/in_progress pair is replayed ahead of the error and the client sees the
announcement the failure belongs to, the same as when no fallback was attempted at all."""
router: Final = _make_router_with_fallback()
trigger: Final = MidStreamFallbackError(
message="dropped before output", model="gpt-4", llm_provider="openai", is_pre_first_chunk=True
)
held: Final = (MagicMock(type="response.created"), MagicMock(type="response.in_progress"))
fallback_error: Final = RuntimeError("fallback closed before its first event")
adopt_headers: Final = MagicMock(return_value=({}, {}))
with patch.object(
router,
"async_function_with_fallbacks_common_utils",
return_value=_make_responses_iterator(error=fallback_error),
):
outcome: Final = [
item
async for item in _events_until_error(
router._aresponses_fallback_attempt(
trigger,
_make_responses_iterator(),
{"model": "gpt-4", "stream": True, "input": "Hello"},
adopt_headers,
held,
)
)
]
assert outcome == [*held, fallback_error]
@pytest.mark.asyncio
async def test_aresponses_fallback_attempt_does_not_replay_held_lifecycle_events_once_the_fallback_announced_itself():
"""Once the fallback has yielded its own created event, a later failure must not replay the
primary's held pair on top of it, or the client would again see two announced response ids."""
router: Final = _make_router_with_fallback()
trigger: Final = MidStreamFallbackError(
message="dropped before output", model="gpt-4", llm_provider="openai", is_pre_first_chunk=True
)
held: Final = (MagicMock(type="response.created"), MagicMock(type="response.in_progress"))
fallback_created: Final = MagicMock(type="response.created")
fallback_error: Final = RuntimeError("fallback dropped after announcing itself")
adopt_headers: Final = MagicMock(return_value=({}, {}))
with patch.object(
router,
"async_function_with_fallbacks_common_utils",
return_value=_make_responses_iterator(chunks=(fallback_created,), error=fallback_error),
):
outcome: Final = [
item
async for item in _events_until_error(
router._aresponses_fallback_attempt(
trigger,
_make_responses_iterator(),
{"model": "gpt-4", "stream": True, "input": "Hello"},
adopt_headers,
held,
)
)
]
assert outcome == [fallback_created, fallback_error]
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_partial_content_injects_continuation():
"""Mid-stream error: input is rewritten to include user prompt +