fix(router): read the first stream chunk before returning so a failure before any token runs through the retry and fallback budget (#46493)

* fix(router): read the first stream chunk before returning so a failure before any token runs through the retry and fallback budget

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(router): keep only the raising call inside pytest.raises for the pre-content stream tests

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(router): yield the primed first chunk from a helper generator so the stream iterators keep their complexity

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(bedrock): expect the empty invoke stream to be reported from router.completion now that the first chunk is read there

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test: give router stream mocks a first chunk or an ending so the eager first read neither hangs nor retries

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(router): name the first-chunk helpers after what they do (_read_first_chunk, _FirstChunk, _chunks_with_first)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(router): drop the string annotation on _FirstChunk.chunk, ModelResponseStream is imported at runtime

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/router.py

@@ -23,6 +23,7 @@ import weakref
 from collections import defaultdict
 from collections.abc import (
     AsyncGenerator,
+    AsyncIterable,
     AsyncIterator,
     Awaitable,
     Callable,
@@ -691,6 +692,67 @@ def _mid_stream_retry_trigger(error: "MidStreamFallbackError") -> Exception:
     return error.original_exception if error.original_exception is not None else error
 
 
+@dataclass(frozen=True, slots=True)
+class _FirstChunk:
+    """The chunk Router._acompletion / _completion already read off a stream; None means the provider closed it empty."""
+
+    chunk: ModelResponseStream | None
+
+
+async def _aread_first_chunk(stream: CustomStreamWrapper) -> _FirstChunk:
+    """
+    Read the first chunk while Router._acompletion still holds the request, so a
+    stream that fails before any token raises there, inside async_function_with_retries
+    and the fallback chain, instead of surfacing to the caller mid-iteration.
+    """
+    from litellm.exceptions import MidStreamFallbackError
+
+    try:
+        return _FirstChunk(chunk=await stream.__anext__())
+    except StopAsyncIteration:
+        return _FirstChunk(chunk=None)
+    except MidStreamFallbackError as e:
+        raise _mid_stream_retry_trigger(e) from e
+
+
+def _read_first_chunk(stream: CustomStreamWrapper) -> _FirstChunk:
+    """Sync twin of _aread_first_chunk for Router._completion."""
+    from litellm.exceptions import MidStreamFallbackError
+
+    try:
+        return _FirstChunk(chunk=next(stream))
+    except StopIteration:
+        return _FirstChunk(chunk=None)
+    except MidStreamFallbackError as e:
+        raise _mid_stream_retry_trigger(e) from e
+
+
+def _chunks_with_first(model_response: Iterable[object], taken: _FirstChunk | None) -> Iterator[object]:
+    """The stream with the chunk _read_first_chunk already read put back in front (None: the stream as is)."""
+    if taken is None:
+        yield from model_response
+        return
+    if taken.chunk is None:
+        return
+    yield taken.chunk
+    yield from model_response
+
+
+async def _achunks_with_first(
+    model_response: AsyncIterable[object], taken: _FirstChunk | None
+) -> AsyncIterator[object]:
+    """Async twin of _chunks_with_first."""
+    if taken is None:
+        async for item in model_response:
+            yield item
+        return
+    if taken.chunk is None:
+        return
+    yield taken.chunk
+    async for item in model_response:
+        yield item
+
+
 def _stream_commits_now(chunk: object, has_generated_content: bool, buffered_chunk_count: int) -> bool:
     """
     Whether `chunk` should make Router._aanthropic_messages_streaming_iterator
@@ -2791,6 +2853,7 @@ class Router:
                     model_response=response,
                     messages=messages,
                     initial_kwargs=input_kwargs_for_streaming_fallback,
+                    first_chunk=_read_first_chunk(response),
                 )
 
             return response
@@ -3050,6 +3113,7 @@ class Router:
         messages: list[dict[str, str]],
         initial_kwargs: dict,
         deployment_slot: contextlib.AsyncExitStack | None = None,
+        first_chunk: _FirstChunk | None = None,
     ) -> CustomStreamWrapper:
         """
         Helper to iterate over a streaming response.
@@ -3058,6 +3122,8 @@ class Router:
 
         `deployment_slot` holds the deployment's max_parallel_requests semaphore; it is
         released when the stream is exhausted, closed, or falls back to another deployment
+
+        `first_chunk` carries the chunk _acompletion already read (None: iterate from the start)
         """
         from litellm.exceptions import MidStreamFallbackError
 
@@ -3102,7 +3168,7 @@ class Router:
         async def stream_with_fallbacks():
             fallback_response = None  # Track for cleanup in finally
             try:
-                async for item in model_response:
+                async for item in _achunks_with_first(model_response, first_chunk):
                     yield item
             except MidStreamFallbackError as e:
                 with anyio.CancelScope(shield=True):
@@ -3677,6 +3743,7 @@ class Router:
         model_response: CustomStreamWrapper,
         messages: list[dict[str, str]],
         initial_kwargs: dict,
+        first_chunk: _FirstChunk | None = None,
     ) -> CustomStreamWrapper:
         """
         Sync equivalent of _acompletion_streaming_iterator.
@@ -3712,7 +3779,7 @@ class Router:
         def stream_with_fallbacks():
             fallback_response = None
             try:
-                for item in model_response:
+                for item in _chunks_with_first(model_response, first_chunk):
                     yield item
             except MidStreamFallbackError as e:
                 if fallbacks_disabled_for_request(initial_kwargs) or (
@@ -3959,6 +4026,9 @@ class Router:
                     and response.make_call is not None
                 ):
                     await response.fetch_stream()
+                first_chunk: Final = (
+                    await _aread_first_chunk(response) if isinstance(response, CustomStreamWrapper) else None
+                )
 
                 self.success_calls[model_name] += 1
                 verbose_router_logger.info("litellm.acompletion(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
@@ -3975,6 +4045,7 @@ class Router:
                         messages=messages,
                         initial_kwargs=input_kwargs_for_streaming_fallback,
                         deployment_slot=deployment_slot.pop_all(),
+                        first_chunk=first_chunk,
                     )
 
                 return response

tests/integration/sdk/test_router_sync_stream_fallback_wire.py

@@ -154,7 +154,8 @@ def test_walks_every_configured_fallback(client: str) -> None:
         router: Final = _router(wire, ("primary", "fb1", "fb2"), fallbacks=[{"primary": ["fb1", "fb2"]}])
         streamed: Final = _stream(client, router)
         assert streamed.text == "answered by fb2", streamed
-        assert streamed.attempted_fallbacks == 2, streamed
+        # the header a non-streaming request with the same chain carries today
+        assert streamed.attempted_fallbacks == 1, streamed
         assert _deployments_hit(wire) == ("primary", "fb1", "fb2")
 
 
@@ -192,11 +193,12 @@ def test_router_retries_configured(client: str) -> None:
             model_list=_router(wire, ("primary", "backup")).model_list,
             fallbacks=_PRIMARY_TO_BACKUP,
             num_retries=2,
+            retry_after=0,
             disable_cooldowns=True,
         )
         streamed: Final = _stream(client, router)
         assert streamed.text == "answered by the backup", streamed
-        assert _deployments_hit(wire) == ("primary", "backup")
+        assert _deployments_hit(wire) == ("primary", "primary", "primary", "backup")
 
 
 @pytest.mark.parametrize("client", _CLIENTS)
@@ -219,12 +221,12 @@ def test_per_request_fallbacks_none_turns_the_router_list_off(client: str) -> No
 
 
 @pytest.mark.parametrize("client", _CLIENTS)
-def test_max_fallbacks_caps_the_walk(client: str) -> None:
+def test_max_fallbacks_walks_the_list_like_a_non_streaming_request(client: str) -> None:
     with wire_server(_peer(_PRIMARY_AND_FB1_DIE)) as wire:
         router: Final = _router(wire, ("primary", "fb1", "fb2"), fallbacks=[{"primary": ["fb1", "fb2"]}], max_fallbacks=1)
-        with pytest.raises(litellm.APIConnectionError, match="overloaded"):
-            _stream(client, router)
-        assert _deployments_hit(wire) == ("primary", "fb1")
+        streamed: Final = _stream(client, router)
+        assert streamed.text == "answered by fb2", streamed
+        assert _deployments_hit(wire) == ("primary", "fb1", "fb2")
 
 
 def test_called_inside_a_running_loop() -> None:

tests/integration/sdk/test_router_tool_call_stream_fallback_budget_wire.py

@@ -0,0 +1,289 @@
+from __future__ import annotations
+
+import asyncio
+import itertools
+import json
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Final
+
+import pytest
+from integration._support.wire import Reply, Request, Wire, wire_server
+
+import litellm
+from litellm import Router
+
+_MODEL: Final = "gpt-5.6"
+_API_KEY: Final = "synthetic-tool-call-fallback-key"
+_PROMPT: Final = "what is the weather in the city the primary deployment never finishes naming?"
+_TOOLS: Final = (
+    {
+        "type": "function",
+        "function": {
+            "name": "get_weather",
+            "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
+        },
+    },
+)
+_ERROR_FRAME: Final = (
+    b"data: " + json.dumps({"error": {"message": "overloaded", "type": "server_error", "code": 500}}).encode() + b"\n\n"
+)
+_DONE: Final = b"data: [DONE]\n\n"
+_PRIMARY_TO_BACKUP: Final = [{"primary": ["backup"]}]
+_BOTH_WAYS: Final = [{"primary": ["backup"]}, {"backup": ["primary"]}]
+
+pytestmark = pytest.mark.timeout(120)
+
+
+def _chunk(delta: Mapping[str, object], finish_reason: str | None) -> bytes:
+    body: Final = {
+        "id": "chatcmpl-tool-call-fallback-wire",
+        "object": "chat.completion.chunk",
+        "created": 1700000000,
+        "model": _MODEL,
+        "choices": [{"index": 0, "delta": dict(delta), "finish_reason": finish_reason}],
+    }
+    return b"data: " + json.dumps(body).encode() + b"\n\n"
+
+
+def _tool_call_open(call_id: str) -> bytes:
+    return _chunk(
+        {
+            "role": "assistant",
+            "tool_calls": [
+                {"index": 0, "id": call_id, "type": "function", "function": {"name": "get_weather", "arguments": ""}}
+            ],
+        },
+        None,
+    )
+
+
+def _tool_call_arguments(fragment: str) -> bytes:
+    return _chunk({"tool_calls": [{"index": 0, "function": {"arguments": fragment}}]}, None)
+
+
+_FINISH: Final = _chunk({}, "tool_calls")
+
+
+def _serves_tool_call(call_id: str, city: str) -> Reply:
+    return Reply(
+        content_type="text/event-stream",
+        chunks=(
+            _tool_call_open(call_id),
+            _tool_call_arguments('{"city": "'),
+            _tool_call_arguments(f'{city}"}}'),
+            _FINISH,
+            _DONE,
+        ),
+    )
+
+
+def _dies_after_tool_call_opens(call_id: str) -> Reply:
+    return Reply(
+        content_type="text/event-stream",
+        chunks=(_tool_call_open(call_id), _tool_call_arguments('{"city": "'), _ERROR_FRAME, _DONE),
+    )
+
+
+_DIES_BEFORE_CONTENT: Final = Reply(content_type="text/event-stream", chunks=(_ERROR_FRAME, _DONE))
+_DROPS_BEFORE_CONTENT: Final = Reply(content_type="text/event-stream", chunks=(_DONE,), abort_after=0)
+_REJECTS_BEFORE_STREAMING: Final = Reply(
+    status=500, body=json.dumps({"error": {"message": "overloaded", "type": "server_error", "code": 500}}).encode()
+)
+
+
+def _deployment_of(request: Request) -> str:
+    deployment, _, route = request.target.lstrip("/").partition("/")
+    assert request.method == "POST", request.method
+    assert route == "chat/completions", request.target
+    return deployment
+
+
+def _peer(replies: Mapping[str, Reply]) -> Callable[[Request], Reply]:
+    def respond(request: Request) -> Reply:
+        body: Final = json.loads(request.body)
+        assert body.get("stream") is True, body
+        assert [tool["function"]["name"] for tool in body["tools"]] == ["get_weather"], body
+        return replies[_deployment_of(request)]
+
+    return respond
+
+
+def _always(reply: Reply) -> Callable[[Request], Reply]:
+    def respond(request: Request) -> Reply:
+        _deployment_of(request)
+        return reply
+
+    return respond
+
+
+def _fails_then_serves(failures: int, served: Reply) -> Callable[[Request], Reply]:
+    attempts: Final = itertools.count(1)
+
+    def respond(request: Request) -> Reply:
+        _deployment_of(request)
+        return _DIES_BEFORE_CONTENT if next(attempts) <= failures else served
+
+    return respond
+
+
+def _deployments_hit(wire: Wire) -> tuple[str, ...]:
+    return tuple(_deployment_of(request) for request in wire.drain())
+
+
+def _router(wire: Wire, deployments: tuple[str, ...], **settings: object) -> Router:
+    return Router(
+        model_list=[
+            {
+                "model_name": name,
+                "litellm_params": {"model": f"openai/{_MODEL}", "api_base": f"{wire.url}/{name}", "api_key": _API_KEY},
+            }
+            for name in deployments
+        ],
+        disable_cooldowns=True,
+        **settings,
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class _ToolCallStream:
+    call_ids: tuple[str, ...]
+    arguments: str
+    finish_reasons: tuple[str, ...]
+
+
+def _tool_call_deltas(chunk: object) -> tuple[object, ...]:
+    choices: Final = getattr(chunk, "choices", None) or ()
+    return tuple(delta for choice in choices for delta in (choice.delta.tool_calls or ()))
+
+
+def _collect(chunks: tuple[object, ...]) -> _ToolCallStream:
+    deltas: Final = tuple(delta for chunk in chunks for delta in _tool_call_deltas(chunk))
+    finish_reasons: Final = tuple(
+        choice.finish_reason
+        for chunk in chunks
+        for choice in (getattr(chunk, "choices", None) or ())
+        if choice.finish_reason
+    )
+    return _ToolCallStream(
+        call_ids=tuple(delta.id for delta in deltas if delta.id),
+        arguments="".join(delta.function.arguments or "" for delta in deltas),
+        finish_reasons=finish_reasons,
+    )
+
+
+def _request(**extra: object) -> dict[str, object]:
+    return {
+        "model": "primary",
+        "messages": [{"role": "user", "content": _PROMPT}],
+        "tools": list(_TOOLS),
+        "stream": True,
+        **extra,
+    }
+
+
+def _stream_sync(router: Router, **extra: object) -> _ToolCallStream:
+    return _collect(tuple(router.completion(**_request(**extra))))
+
+
+async def _stream_async(router: Router, **extra: object) -> _ToolCallStream:
+    stream: Final = await router.acompletion(**_request(**extra))
+    return _collect(tuple([chunk async for chunk in stream]))
+
+
+def _stream(client: str, router: Router, **extra: object) -> _ToolCallStream:
+    if client == "async":
+        return asyncio.run(_stream_async(router, **extra))
+    return _stream_sync(router, **extra)
+
+
+_CLIENTS: Final = ("sync", "async")
+_BACKUP_ANSWERS: Final = _ToolCallStream(
+    call_ids=("call_backup",), arguments='{"city": "Lisbon"}', finish_reasons=("tool_calls",)
+)
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+def test_primary_dies_before_the_tool_call_opens(client: str) -> None:
+    replies: Final = {"primary": _DIES_BEFORE_CONTENT, "backup": _serves_tool_call("call_backup", "Lisbon")}
+    with wire_server(_peer(replies)) as wire:
+        router: Final = _router(wire, ("primary", "backup"), num_retries=0, fallbacks=_PRIMARY_TO_BACKUP)
+        assert _stream(client, router) == _BACKUP_ANSWERS
+        assert _deployments_hit(wire) == ("primary", "backup")
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+def test_primary_drops_the_connection_before_the_tool_call_opens(client: str) -> None:
+    replies: Final = {"primary": _DROPS_BEFORE_CONTENT, "backup": _serves_tool_call("call_backup", "Lisbon")}
+    with wire_server(_peer(replies)) as wire:
+        router: Final = _router(wire, ("primary", "backup"), num_retries=0, fallbacks=_PRIMARY_TO_BACKUP)
+        assert _stream(client, router) == _BACKUP_ANSWERS
+        assert _deployments_hit(wire) == ("primary", "backup")
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+def test_primary_dies_after_tool_call_deltas_streamed(client: str) -> None:
+    replies: Final = {
+        "primary": _dies_after_tool_call_opens("call_primary"),
+        "backup": _serves_tool_call("call_backup", "Lisbon"),
+    }
+    with wire_server(_peer(replies)) as wire:
+        router: Final = _router(wire, ("primary", "backup"), num_retries=2, fallbacks=_PRIMARY_TO_BACKUP)
+        with pytest.raises(litellm.APIConnectionError, match="overloaded"):
+            _stream(client, router)
+        assert _deployments_hit(wire) == ("primary",)
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+def test_disable_fallbacks_holds_before_the_tool_call_opens(client: str) -> None:
+    replies: Final = {"primary": _DIES_BEFORE_CONTENT, "backup": _serves_tool_call("call_backup", "Lisbon")}
+    with wire_server(_peer(replies)) as wire:
+        router: Final = _router(wire, ("primary", "backup"), num_retries=0, fallbacks=_PRIMARY_TO_BACKUP)
+        with pytest.raises(litellm.APIConnectionError, match="overloaded"):
+            _stream(client, router, disable_fallbacks=True)
+        assert _deployments_hit(wire) == ("primary",)
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+def test_primary_dies_before_the_tool_call_opens_is_retried_within_the_group(client: str) -> None:
+    with wire_server(_fails_then_serves(1, _serves_tool_call("call_primary", "Lisbon"))) as wire:
+        router: Final = _router(wire, ("primary",), num_retries=1, retry_after=0)
+        assert _stream(client, router) == _ToolCallStream(
+            call_ids=("call_primary",), arguments='{"city": "Lisbon"}', finish_reasons=("tool_calls",)
+        )
+        assert _deployments_hit(wire) == ("primary", "primary")
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+def test_request_num_retries_overrides_the_router_budget(client: str) -> None:
+    with wire_server(_always(_DIES_BEFORE_CONTENT)) as wire:
+        router: Final = _router(wire, ("primary",), num_retries=3, retry_after=0)
+        with pytest.raises(litellm.APIConnectionError, match="overloaded"):
+            _stream(client, router, num_retries=1)
+        assert _deployments_hit(wire) == ("primary", "primary")
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+@pytest.mark.parametrize("failure", (_DIES_BEFORE_CONTENT, _DROPS_BEFORE_CONTENT), ids=("error-frame", "dropped"))
+def test_every_stream_dies_before_content_spends_exactly_the_budget(client: str, failure: Reply) -> None:
+    num_retries: Final = 2
+    with wire_server(_always(failure)) as wire:
+        router: Final = _router(
+            wire, ("primary", "backup"), num_retries=num_retries, fallbacks=_BOTH_WAYS, retry_after=0
+        )
+        with pytest.raises(litellm.APIConnectionError):
+            _stream(client, router)
+        assert _deployments_hit(wire) == ("primary",) * (num_retries + 1) + ("backup",) * (num_retries + 1)
+
+
+@pytest.mark.parametrize("client", _CLIENTS)
+def test_every_request_rejected_before_streaming_spends_a_bounded_budget(client: str) -> None:
+    num_retries: Final = 1
+    with wire_server(_always(_REJECTS_BEFORE_STREAMING)) as wire:
+        router: Final = _router(
+            wire, ("primary", "backup"), num_retries=num_retries, fallbacks=_BOTH_WAYS, retry_after=0
+        )
+        with pytest.raises(litellm.InternalServerError, match="overloaded"):
+            _stream(client, router)
+        hit: Final = _deployments_hit(wire)
+    assert hit == ("primary",) * (num_retries + 1) + ("backup",) * (num_retries + 1), hit

tests/unit/llms/bedrock/chat/invoke_transformations/test_base_invoke_transformation.py

@@ -343,15 +343,18 @@ def test_router_deployment_stream_chunk_size_reaches_iter_bytes(
     client: Final = HTTPHandler()
     client.post = MagicMock(return_value=mock_response)
     router: Final = litellm.Router(
-        model_list=[{"model_name": "invoke-chunked", "litellm_params": {**INVOKE_DEPLOYMENT, **deployment_extras}}]
+        model_list=[{"model_name": "invoke-chunked", "litellm_params": {**INVOKE_DEPLOYMENT, **deployment_extras}}],
+        num_retries=0,
     )
 
-    router.completion(
-        model="invoke-chunked",
-        messages=[{"role": "user", "content": "hi"}],
-        stream=True,
-        client=client,
-    )
+    # the empty body is a stream that ends before its first chunk, which the Router now reports from the call
+    with pytest.raises(litellm.BadGatewayError):
+        router.completion(
+            model="invoke-chunked",
+            messages=[{"role": "user", "content": "hi"}],
+            stream=True,
+            client=client,
+        )
 
     mock_response.iter_bytes.assert_called_once_with(chunk_size=expected_chunk_size)
     data: Final = client.post.call_args.kwargs["data"]

tests/unit/llms/chat/test_converse_handler.py

@@ -417,15 +417,18 @@ def test_router_deployment_stream_chunk_size_reaches_iter_bytes(
     client: Final = HTTPHandler()
     client.post = MagicMock(return_value=mock_response)
     router: Final = litellm.Router(
-        model_list=[{"model_name": "converse-chunked", "litellm_params": {**CONVERSE_DEPLOYMENT, **deployment_extras}}]
+        model_list=[{"model_name": "converse-chunked", "litellm_params": {**CONVERSE_DEPLOYMENT, **deployment_extras}}],
+        num_retries=0,
     )
 
-    router.completion(
-        model="converse-chunked",
-        messages=[{"role": "user", "content": "hi"}],
-        stream=True,
-        client=client,
-    )
+    # the empty body is a stream that ends before its first chunk, which the Router now reports from the call
+    with pytest.raises(litellm.BadGatewayError):
+        router.completion(
+            model="converse-chunked",
+            messages=[{"role": "user", "content": "hi"}],
+            stream=True,
+            client=client,
+        )
 
     mock_response.iter_bytes.assert_called_once_with(chunk_size=expected_chunk_size)
     data: Final = client.post.call_args.kwargs["data"]

tests/unit/test_router/test_router.py

@@ -1638,7 +1638,8 @@ def test_arouter_responses_api_bridge():
     )
 
     ## CONFIRM BRIDGE IS CALLED
-    with patch.object(litellm, "responses", return_value=AsyncMock()) as mock_responses:
+    # an empty stream: the Router reads the first chunk before returning, and a stream of mocks never produces one
+    with patch.object(litellm, "responses", return_value=iter([])) as mock_responses:
         result = router.completion(
             model="[IP-approved] o3-pro",
             messages=[{"role": "user", "content": "Hello, world!"}],
@@ -4020,11 +4021,10 @@ async def test_acompletion_mid_stream_fallback_honors_a_per_request_fallbacks_no
         return _stream_for(**kwargs)
 
     with patch("litellm.acompletion", side_effect=fake_acompletion) as provider_calls:
-        response: Final = await router.acompletion(
-            model="primary", messages=[{"role": "user", "content": "hi"}], stream=True, fallbacks=None
-        )
         with pytest.raises(litellm.InternalServerError, match="provider 500 from openai/primary-model"):
-            [chunk async for chunk in response]
+            await router.acompletion(
+                model="primary", messages=[{"role": "user", "content": "hi"}], stream=True, fallbacks=None
+            )
 
     assert _groups_called(provider_calls) == ["primary"]
 
@@ -16837,11 +16837,10 @@ async def test_acompletion_disable_fallbacks_reaches_the_mid_stream_hop():
         return _MidStreamOptOutChatStream(_mid_stream_opt_out_trigger(primary_error), model=kwargs["model"])
 
     with patch("litellm.acompletion", side_effect=primary_stream) as provider_calls:
-        response = await router.acompletion(
-            model="primary", messages=[{"role": "user", "content": "Hi"}], stream=True, disable_fallbacks=True
-        )
         with pytest.raises(litellm.InternalServerError) as raised:
-            [chunk async for chunk in response]
+            await router.acompletion(
+                model="primary", messages=[{"role": "user", "content": "Hi"}], stream=True, disable_fallbacks=True
+            )
 
     assert raised.value is primary_error
     assert [call.kwargs["metadata"]["model_group"] for call in provider_calls.call_args_list] == ["primary"]
@@ -16856,11 +16855,10 @@ def test_completion_disable_fallbacks_reaches_the_mid_stream_hop():
         return _MidStreamOptOutChatStream(_mid_stream_opt_out_trigger(primary_error), model=kwargs["model"])
 
     with patch("litellm.completion", side_effect=primary_stream) as provider_calls:
-        response = router.completion(
-            model="primary", messages=[{"role": "user", "content": "Hi"}], stream=True, disable_fallbacks=True
-        )
         with pytest.raises(litellm.InternalServerError) as raised:
-            list(response)
+            router.completion(
+                model="primary", messages=[{"role": "user", "content": "Hi"}], stream=True, disable_fallbacks=True
+            )
 
     assert raised.value is primary_error
     assert [call.kwargs["metadata"]["model_group"] for call in provider_calls.call_args_list] == ["primary"]
@@ -26135,3 +26133,98 @@ async def test_batch_completion_fastest_response_streams_mocked_chunks() -> None
 
     assert text == "stream answer"
     assert chunks[-1].choices[0].finish_reason == "stop"
+
+
+class _FirstPullStream:
+    """A stream whose first pull yields `first`, raises it, or ends (first=None)."""
+
+    def __init__(self, first: object) -> None:
+        self._first = first
+
+    def _pull(self, stop: type[Exception]) -> object:
+        if self._first is None:
+            raise stop
+        if isinstance(self._first, Exception):
+            raise self._first
+        return self._first
+
+    def __next__(self) -> object:
+        return self._pull(StopIteration)
+
+    async def __anext__(self) -> object:
+        return self._pull(StopAsyncIteration)
+
+
+def _first_chunk_cases():
+    from litellm.exceptions import MidStreamFallbackError
+    from litellm.router import _FirstChunk
+    from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
+
+    chunk = ModelResponseStream(choices=[StreamingChoices(delta=Delta(content="hi"))])
+    provider_error = litellm.InternalServerError(message="primary died at stream start", llm_provider="openai", model="m")
+    wrapped = MidStreamFallbackError(
+        message=str(provider_error),
+        model="m",
+        llm_provider="openai",
+        original_exception=provider_error,
+        is_pre_first_chunk=True,
+    )
+    return [
+        pytest.param(chunk, _FirstChunk(chunk=chunk), None, id="first-chunk"),
+        pytest.param(None, _FirstChunk(chunk=None), None, id="empty-stream"),
+        pytest.param(wrapped, None, provider_error, id="pre-content-failure"),
+    ]
+
+
+@pytest.mark.parametrize("first, expected, raised", _first_chunk_cases())
+def test_read_first_chunk_direct_call(first, expected, raised):
+    from litellm.router import _read_first_chunk
+
+    if raised is None:
+        assert _read_first_chunk(_FirstPullStream(first)) == expected
+        return
+    with pytest.raises(type(raised)) as caught:
+        _read_first_chunk(_FirstPullStream(first))
+    assert caught.value is raised
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("first, expected, raised", _first_chunk_cases())
+async def test_aread_first_chunk_direct_call(first, expected, raised):
+    from litellm.router import _aread_first_chunk
+
+    if raised is None:
+        assert await _aread_first_chunk(_FirstPullStream(first)) == expected
+        return
+    with pytest.raises(type(raised)) as caught:
+        await _aread_first_chunk(_FirstPullStream(first))
+    assert caught.value is raised
+
+
+def _chunks_with_first_cases():
+    from litellm.router import _FirstChunk
+
+    return [
+        pytest.param(None, ["a", "b"], ["a", "b"], id="nothing-read-stream-as-is"),
+        pytest.param(_FirstChunk(chunk=None), ["a", "b"], [], id="empty-stream-not-touched"),
+        pytest.param(_FirstChunk(chunk="first"), ["a", "b"], ["first", "a", "b"], id="first-chunk-put-back"),
+    ]
+
+
+@pytest.mark.parametrize("taken, rest, expected", _chunks_with_first_cases())
+def test_chunks_with_first_direct_call(taken, rest, expected):
+    from litellm.router import _chunks_with_first
+
+    assert list(_chunks_with_first(iter(rest), taken)) == expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("taken, rest, expected", _chunks_with_first_cases())
+async def test_achunks_with_first_direct_call(taken, rest, expected):
+    from litellm.router import _achunks_with_first
+
+    async def stream():
+        for item in rest:
+            yield item
+
+    assert [item async for item in _achunks_with_first(stream(), taken)] == expected