fix(router): fall back from unhealthy auto-router tier

Co-Authored-By: Claude Code <noreply@anthropic.com>

litellm/router.py

@@ -12669,6 +12669,7 @@ class Router:
         input: str | list | None = None,
         specific_deployment: bool | None = False,
         parent_otel_span: Span | None = None,
+        health_check_probe: bool = False,
     ) -> list[dict] | dict:
         """
         Get the healthy deployments for a model.
@@ -12718,6 +12719,7 @@ class Router:
         healthy_deployments = await self._async_filter_health_check_unhealthy_deployments(
             healthy_deployments=healthy_deployments,
             parent_otel_span=parent_otel_span,
+            health_check_probe=health_check_probe,
         )
 
         cooldown_deployments: Final = await _async_get_cooldown_deployments(
@@ -14100,6 +14102,7 @@ class Router:
         self,
         healthy_deployments: list[dict],
         parent_otel_span: Span | None = None,
+        health_check_probe: bool = False,
     ) -> list[dict]:
         """
         Filter out deployments marked unhealthy by background health checks.
@@ -14136,8 +14139,7 @@ class Router:
         ]
 
         if not filtered:
-            verbose_router_logger.warning("All deployments marked unhealthy by health checks, bypassing health filter")
-            return healthy_deployments
+            return [] if health_check_probe else healthy_deployments  # mutable-ok: empty list signals unavailable probe
 
         return filtered
 

litellm/router_strategy/adaptive_router/adaptive_router.py

@@ -461,6 +461,8 @@ class AdaptiveRouter:
                 if d_alpha == 0 and d_beta == 0:
                     continue
                 cell_key = (attribution_type, target_model)
+                if cell_key not in self._cells:
+                    continue
                 self._cells[cell_key] = apply_delta(
                     self._cells[cell_key],
                     d_alpha,

litellm/router_strategy/complexity_router/README.md

@@ -270,6 +270,18 @@ change or default takeover records `cause: modality_escalation` with the displac
 pinned by session affinity, and by default a KEPT session pin bypasses the gate: a session pinned
 to a text-only model keeps it even when an image arrives.
 
+Context-window and modality recovery take priority over the default model. If a compatible tier
+cannot serve, the router checks the remaining compatible recovery tiers before using `default_model`.
+A capacity failure without those constraints tries the selected tier's peers, then the default
+
+The default must fit the context and accept the request's modality. It cannot bypass routing plugins
+or a plan-mode floor. Context fit uses the auto-router's existing buffer even when Router-wide pre-call
+checks are off. Missing context metadata retains the existing unknown-window behavior
+
+Health fallback records `cause: health_default_fallback` and `health_displaced:<MODEL>` in `signals`.
+It does not replace the session's tier pin. Adaptive feedback retains the model that actually served,
+but a default outside the adaptive candidate pool does not become a normal candidate
+
 Add `modality_pin_override: true` to lift that last exemption. The image turn is then re-placed
 the same way every other decision is, and records `cause: modality_pin_override` whether or not
 the tier moved, since the model left the pin either way. The pin itself is untouched: the session

litellm/router_strategy/complexity_router/complexity_router.py

@@ -936,6 +936,7 @@ def _decision_is_pinnable(decision: StandardLoggingRoutingDecision | None) -> bo
             "modality_escalation",
             "modality_pin_override",
             "health_failover",
+            "health_default_fallback",
         )
         and not decision.get("context_escalated")
         and _CLASSIFIER_CIRCUIT_OPEN_SIGNAL not in (decision.get("signals") or ())
@@ -1097,6 +1098,15 @@ def _group_provably_fits(facts: tuple[int | None, bool], needed: int, buffer: fl
     return window is not None and not has_unknown and needed <= int(window * buffer)
 
 
+class _RequestContextFit(NamedTuple):
+    facts: Mapping[str, tuple[int | None, bool]]
+    needed: int | None
+    buffer: float
+
+    def accepts(self, model: str) -> bool:
+        return self.needed is None or _window_can_hold(self.facts.get(model, (None, True))[0], self.needed, self.buffer)
+
+
 class _ContextWindowPlacement(NamedTuple):
     """Where the context-window gate placed the request: the placement tier, the subset of its
     pool the pick may use, and every configured group not provably misfit (the adaptive filter)."""
@@ -2672,12 +2682,32 @@ class ComplexityRouter(CustomLogger):
             verbose_router_logger.debug("ComplexityRouter: context-window token count failed. Got - %s", e)
             return None
 
+    async def _request_context_fit(
+        self,
+        resolved_messages: Sequence[Mapping[str, object]] | None,
+        request_kwargs: Mapping[str, object],
+    ) -> _RequestContextFit:
+        if not self.config.enable_context_window_escalation or not resolved_messages:
+            return _RequestContextFit(EMPTY_MAPPING, None, self.config.context_window_escalation_buffer)
+        names: Final = frozenset(model for pool in self._tier_pools().values() for model in pool) | frozenset(
+            (self.config.default_model,) if self.config.default_model else ()
+        )
+        facts: Final = MappingProxyType({name: self._group_window_facts(name) for name in names})
+        known: Final = tuple(window for window, _ in facts.values() if window is not None)
+        buffer: Final = self.config.context_window_escalation_buffer
+        needs_count: Final = known and self._request_byte_upper_bound(resolved_messages, request_kwargs) > int(
+            min(known) * buffer
+        )
+        needed: Final = await self._counted_request_tokens(resolved_messages, request_kwargs) if needs_count else None
+        return _RequestContextFit(facts=facts, needed=needed, buffer=buffer)
+
     async def _context_window_placement(
         self,
         tier: ComplexityTier | str,
         resolved_messages: Sequence[Mapping[str, object]] | None,
         request_kwargs: Mapping[str, object],
         pool_override: tuple[str, ...] | None = None,
+        context_fit: _RequestContextFit | None = None,
     ) -> _ContextWindowPlacement | None:
         """Correct a decided placement whose models provably cannot hold the prompt, or None
         (the placement stands). Only a real tokenizer count ever moves a request, escalation
@@ -2689,17 +2719,10 @@ class ComplexityRouter(CustomLogger):
         pool: Final = pool_override if pool_override is not None else tuple(pools.get(_tier_name(tier), ()))
         if not pool:
             return None
-        facts: Final = MappingProxyType({group: self._group_window_facts(group) for group in pool})
-        known_windows: Final = tuple(window for window, _ in facts.values() if window is not None)
-        if not known_windows:
+        fit: Final = context_fit or await self._request_context_fit(resolved_messages, request_kwargs)
+        if fit.needed is None:
             return None
-        buffer: Final = self.config.context_window_escalation_buffer
-        if self._request_byte_upper_bound(resolved_messages, request_kwargs) <= int(min(known_windows) * buffer):
-            return None
-        needed: Final = await self._counted_request_tokens(resolved_messages, request_kwargs)
-        if needed is None:
-            return None
-        return self._placement_for_tokens(tier=tier, pool=pool, pools=pools, facts=facts, needed=needed)
+        return self._placement_for_tokens(tier=tier, pool=pool, pools=pools, facts=fit.facts, needed=fit.needed)
 
     def _placement_for_tokens(
         self,
@@ -2711,14 +2734,16 @@ class ComplexityRouter(CustomLogger):
         needed: int,
     ) -> _ContextWindowPlacement | None:
         buffer: Final = self.config.context_window_escalation_buffer
-        in_tier: Final = tuple(group for group in pool if _window_can_hold(facts[group][0], needed, buffer))
+        in_tier: Final = tuple(
+            group for group in pool if _window_can_hold(facts.get(group, (None, True))[0], needed, buffer)
+        )
         if in_tier and len(in_tier) == len(pool):
             return None
         holdable: Final = frozenset(
             group
             for tier_pool in pools.values()
             for group in tier_pool
-            if _window_can_hold(self._group_window_facts(group)[0], needed, buffer)
+            if _window_can_hold(facts.get(group, (None, True))[0], needed, buffer)
         )
         if in_tier:
             return _ContextWindowPlacement(tier=tier, allowed_models=in_tier, holdable_models=holdable)
@@ -2726,7 +2751,7 @@ class ComplexityRouter(CustomLogger):
             proven = tuple(
                 group
                 for group in pools.get(name, ())
-                if _group_provably_fits(self._group_window_facts(group), needed, buffer)
+                if _group_provably_fits(facts.get(group, (None, True)), needed, buffer)
             )
             if proven:
                 return _ContextWindowPlacement(
@@ -2872,6 +2897,7 @@ class ComplexityRouter(CustomLogger):
         messages: list[dict[str, Any]] | None,  # mutable-ok: forwarded verbatim to the list-typed re-pick
         resolved_messages: Sequence[Mapping[str, object]] | None,
         request_kwargs: dict,  # mutable-ok: same shape the hook receives
+        context_fit: _RequestContextFit | None = None,
     ) -> PreRoutingHookResponse:
         """Replace a routed model that cannot accept this request's image input.
 
@@ -2902,7 +2928,8 @@ class ComplexityRouter(CustomLogger):
             or self._model_accepts_image_input(response.model)
         ):
             return response
-        eligible: Final = self._modality_eligible_models()
+        fit: Final = context_fit or await self._request_context_fit(resolved_messages, request_kwargs)
+        eligible: Final = frozenset(name for name in self._modality_eligible_models() if fit.accepts(name))
         names: Final = self.config.tier_names()
         pools: Final = self._tier_pools()
         decided: Final = decision.get("tier") if decision is not None else None
@@ -3036,8 +3063,10 @@ class ComplexityRouter(CustomLogger):
                 messages=messages,
                 input=input,
                 parent_otel_span=_get_parent_otel_span_from_kwargs(request_kwargs),
+                health_check_probe=True,
             )
-        except (RouterRateLimitError, RouterRateLimitErrorBasic, BadRequestError):
+        except (RouterRateLimitError, RouterRateLimitErrorBasic, BadRequestError) as exc:
+            verbose_router_logger.debug("health probe unavailable model=%s error=%s", model_name, type(exc).__name__)
             return False
         except Exception as exc:  # noqa: BLE001  # a speculative eligibility read must fail open on unknown faults
             verbose_router_logger.debug(
@@ -3053,76 +3082,124 @@ class ComplexityRouter(CustomLogger):
         input: str | list | None,  # mutable-ok: mirrors the owner's own input parameter, which this forwards verbatim
         resolved_messages: Sequence[Mapping[str, object]] | None,
         request_kwargs: dict,  # mutable-ok: same shape the hook receives
+        context_fit: _RequestContextFit | None = None,
     ) -> PreRoutingHookResponse:
-        """Replace a decided model group that has no serving capacity with a live peer in the same tier.
-
-        Applied to the decided response at the hook's exits, so every arm that can place a request
-        is covered by one owner: a fresh classification, a replayed or escalated session pin, a
-        plan-mode floor, a context-window escalation, an adaptive pick, and whatever arm is added
-        next. Peers come from the DECIDED tier only; climbing to another tier is deliberately not
-        done here, since a higher tier costs more than the classifier asked for.
-
-        Serving capacity is one question asked of one owner (`_model_group_can_serve`), so the
-        substitute is only ever a group the pipeline would actually accept for this request. The
-        pick then runs through `_pick_model_for_tier`, so routing plugins decide the substitute
-        exactly as they decided the original.
-
-        Fails open everywhere it cannot be sure: an unreadable eligibility view, a decision
-        carrying no tier (default_model), or a tier whose every peer is unusable too. It fails
-        CLOSED on a plugin that empties the pool, leaving the original decision to fail rather
-        than serving a model the plugin excluded.
-        """
+        """Try compatible tier recovery before the default, preserving request policy and fit."""
         decision: Final = response.routing_decision
         decided_tier: Final = decision.get("tier") if decision is not None else None
         if decision is None or not isinstance(decided_tier, str):
             return response
-        peers: Final = tuple(self._tier_pools().get(decided_tier, ()))
-        if len(peers) < 2:
-            return response
-        if await self._model_group_can_serve(response.model, messages, input, request_kwargs):
+        fit: Final = context_fit or await self._request_context_fit(resolved_messages, request_kwargs)
+        if fit.accepts(response.model) and await self._model_group_can_serve(
+            response.model, messages, input, request_kwargs
+        ):
             return response
         eligible: Final = (
             self._modality_eligible_models()
             if self.config.modality_routing and resolved_messages and request_contains_image_content(resolved_messages)
             else None
         )
-        candidates: Final = tuple(
-            peer for peer in peers if peer != response.model and (eligible is None or peer in eligible)
+        pools: Final = self._tier_pools()
+        context_recovery: Final = bool(decision.get("context_escalated")) or any(
+            not fit.accepts(model) for model in pools.get(decided_tier, ())
         )
-        if not candidates:
-            return response
-        servable: Final = await asyncio.gather(
-            *(self._model_group_can_serve(peer, messages, input, request_kwargs) for peer in candidates)
+        modality_recovery: Final = eligible is not None
+        names: Final = self.config.tier_names()
+        tiers: Final = (
+            tuple(names[names.index(decided_tier) :])
+            if (context_recovery or modality_recovery) and decided_tier in names
+            else (decided_tier,)
         )
-        live: Final = tuple(peer for peer, can_serve in zip(candidates, servable) if can_serve)
-        if not live:
-            return response
-        repick_messages: Final = (
-            list(resolved_messages) if resolved_messages else None  # mutable-ok: the pick's param is list-typed
-        )
-        try:
-            new_model: Final = await self._pick_model_for_tier(
-                decided_tier if self.config.has_custom_tiers else ComplexityTier(decided_tier),
-                messages,
-                repick_messages,  # pyright: ignore[reportArgumentType]  # hook-resolved message dicts; the pick only reads them
-                request_kwargs,
-                allowed_models=live,
+
+        async def recover_tier(candidate_tier: str) -> PreRoutingHookResponse | None:
+            peers: Final = tuple(
+                model
+                for model in pools.get(candidate_tier, ())
+                if not context_recovery
+                or candidate_tier == decided_tier
+                or fit.needed is None
+                or _group_provably_fits(fit.facts.get(model, (None, True)), fit.needed, fit.buffer)
             )
-        except ValueError as exc:
-            verbose_router_logger.debug(
-                "ComplexityRouter: health failover found no candidate the routing plugins allow: %s", exc
+            candidates: Final = tuple(
+                peer
+                for peer in peers
+                if peer != response.model and fit.accepts(peer) and (eligible is None or peer in eligible)
             )
+            servable: Final = await asyncio.gather(
+                *(self._model_group_can_serve(peer, messages, input, request_kwargs) for peer in candidates)
+            )
+            live: Final = tuple(peer for peer, can_serve in zip(candidates, servable) if can_serve)
+            if live:
+                repick_messages: Final = (
+                    list(resolved_messages) if resolved_messages else None  # mutable-ok: the pick's param is list-typed
+                )
+                try:
+                    new_model: Final = await self._pick_model_for_tier(
+                        candidate_tier if self.config.has_custom_tiers else ComplexityTier(candidate_tier),
+                        messages,
+                        repick_messages,  # pyright: ignore[reportArgumentType]  # hook-resolved message dicts; the pick only reads them
+                        request_kwargs,
+                        allowed_models=live,
+                    )
+                except ValueError as exc:
+                    verbose_router_logger.debug(
+                        "ComplexityRouter: health failover found no candidate the routing plugins allow: %s", exc
+                    )
+                else:
+                    self._restamp_adaptive_choice(request_kwargs, response.model, new_model)
+                    verbose_router_logger.info(
+                        "ComplexityRouter: routing decision cause=health_failover, routed_model=%s, displaced=%s",
+                        new_model,
+                        response.model,
+                    )
+                    new_decision: Final = self._build_routing_decision(
+                        routed_model=new_model,
+                        cause="health_failover",
+                        tier=candidate_tier,
+                        score=decision.get("score"),
+                        signals=(*(decision.get("signals") or ()), f"health_displaced:{response.model}"),
+                        matched_keyword=decision.get("matched_keyword"),
+                        escalation_keyword=decision.get("escalation_keyword"),
+                        escalated=bool(decision.get("escalated", False)),
+                        classifier_model=decision.get("classifier_model"),
+                        classifier_cost=decision.get("classifier_cost"),
+                        conversation_continuing=bool(decision.get("conversation_continuing", True)),
+                        tier_litellm_params=self._litellm_params_for_model(candidate_tier, new_model),
+                        context_escalation_original_tier=decision.get("context_escalation_original_tier"),
+                    )
+                    return response.model_copy(
+                        update={  # mutable-ok: model_copy types update as a plain dict
+                            "model": new_model,
+                            "litellm_params": self._litellm_params_for_model(candidate_tier, new_model),
+                            "routing_decision": new_decision,
+                        }
+                    )
+            return None
+
+        for candidate_tier in tiers:
+            if (recovered := await recover_tier(candidate_tier)) is not None:
+                return recovered
+        default_model: Final = self.config.default_model
+        plan_mode_active: Final = self._matched_plan_mode_signal(request_kwargs, resolved_messages) is not None
+        if (
+            plan_mode_active
+            or self.config.plugins
+            or not default_model
+            or default_model == response.model
+            or not fit.accepts(default_model)
+            or (eligible is not None and default_model not in eligible)
+            or not await self._model_group_can_serve(default_model, messages, input, request_kwargs)
+        ):
             return response
-        self._restamp_adaptive_choice(request_kwargs, response.model, new_model)
+        self._restamp_adaptive_choice(request_kwargs, response.model, default_model)
         verbose_router_logger.info(
-            "ComplexityRouter: routing decision cause=health_failover, routed_model=%s, displaced=%s",
-            new_model,
+            "ComplexityRouter: routing decision cause=health_default_fallback, routed_model=%s, displaced=%s",
+            default_model,
             response.model,
         )
-        new_decision: Final = self._build_routing_decision(
-            routed_model=new_model,
-            cause="health_failover",
-            tier=decision.get("tier"),
+        default_decision: Final = self._build_routing_decision(
+            routed_model=default_model,
+            cause="health_default_fallback",
             score=decision.get("score"),
             signals=(*(decision.get("signals") or ()), f"health_displaced:{response.model}"),
             matched_keyword=decision.get("matched_keyword"),
@@ -3131,14 +3208,14 @@ class ComplexityRouter(CustomLogger):
             classifier_model=decision.get("classifier_model"),
             classifier_cost=decision.get("classifier_cost"),
             conversation_continuing=bool(decision.get("conversation_continuing", True)),
-            tier_litellm_params=self._litellm_params_for_model(decided_tier, new_model),
+            tier_litellm_params=self._litellm_params_for_model(None, default_model),
             context_escalation_original_tier=decision.get("context_escalation_original_tier"),
         )
         return response.model_copy(
             update={  # mutable-ok: model_copy types update as a plain dict
-                "model": new_model,
-                "litellm_params": self._litellm_params_for_model(decided_tier, new_model),
-                "routing_decision": new_decision,
+                "model": default_model,
+                "litellm_params": self._litellm_params_for_model(None, default_model),
+                "routing_decision": default_decision,
             }
         )
 
@@ -3453,6 +3530,7 @@ class ComplexityRouter(CustomLogger):
         # chat-completions messages, so it is real work on every non-chat surface, and
         # both the conversation shape and the classifier read the same list.
         resolved_messages: Final = self._resolve_messages(messages, request_kwargs)
+        context_fit: Final = await self._request_context_fit(resolved_messages, request_kwargs)
         marker_pairs: Final = self._reminder_markers_for_request(request_kwargs)
         conversation_continuing: Final = _conversation_is_continuing(resolved_messages)
 
@@ -3503,7 +3581,11 @@ class ComplexityRouter(CustomLogger):
                     pin_source_tier: Final = self._tier_for_model(routed_model)
                     pin_placement: Final = (
                         await self._context_window_placement(
-                            pin_source_tier, resolved_messages, request_kwargs, pool_override=(routed_model,)
+                            pin_source_tier,
+                            resolved_messages,
+                            request_kwargs,
+                            pool_override=(routed_model,),
+                            context_fit=context_fit,
                         )
                         if pin_source_tier is not None
                         else None
@@ -3573,11 +3655,13 @@ class ComplexityRouter(CustomLogger):
                                 messages,
                                 resolved_messages,
                                 request_kwargs,
+                                context_fit,
                             ),
                             messages,
                             input,
                             resolved_messages,
                             request_kwargs,
+                            context_fit,
                         )
                     )
 
@@ -3589,14 +3673,18 @@ class ComplexityRouter(CustomLogger):
             specific_deployment=specific_deployment,
             conversation_continuing=conversation_continuing,
             resolved_messages=resolved_messages,
+            context_fit=context_fit,
         )
         response: Final = (
             await self._gate_response_health(
-                await self._gate_response_modality(routed_response, messages, resolved_messages, request_kwargs),
+                await self._gate_response_modality(
+                    routed_response, messages, resolved_messages, request_kwargs, context_fit
+                ),
                 messages,
                 input,
                 resolved_messages,
                 request_kwargs,
+                context_fit,
             )
             if routed_response is not None
             else None
@@ -3631,6 +3719,7 @@ class ComplexityRouter(CustomLogger):
         specific_deployment: bool | None = False,
         conversation_continuing: bool = True,
         resolved_messages: Sequence[Mapping[str, object]] | None = None,
+        context_fit: _RequestContextFit | None = None,
     ) -> PreRoutingHookResponse | None:
         """
         Classifies the request by complexity and returns the appropriate model.
@@ -3802,7 +3891,9 @@ class ComplexityRouter(CustomLogger):
         plan_floored: Final = tier != pre_floor_tier
         if plan_floored:
             signals = (*signals, "plan_mode_floor")
-        context_placement: Final = await self._context_window_placement(tier, resolved_messages, request_kwargs)
+        context_placement: Final = await self._context_window_placement(
+            tier, resolved_messages, request_kwargs, context_fit=context_fit
+        )
         tier, signals, context_original_tier = _apply_context_placement(tier, signals, context_placement)
         score_repr: Final = f"{score:.3f}" if score is not None else "n/a"
         fallback_model: Final = self.config.default_model if not self.config.plugins else None

litellm/types/utils.py

@@ -2920,6 +2920,7 @@ RoutingDecisionCause = Literal[
     # same tier served instead. The displaced group rides in signals. Reported even on a kept
     # session pin, since the pinned model did not serve the request.
     "health_failover",
+    "health_default_fallback",
     "session_affinity_pin",
     "session_affinity_escalation",
     # classification_mode 'user_turn': the request is an agent loop's continuation turn (no new

tests/test_litellm/router_strategy/adaptive_router/test_adaptive_router.py

@@ -236,6 +236,38 @@ async def test_record_turn_attributes_satisfaction_to_previous_response_model():
     assert smart_after.alpha == pytest.approx(smart_before.alpha)
 
 
+@pytest.mark.asyncio
+async def test_external_default_keeps_feedback_history_without_entering_bandit_pool():
+    r = _make_router()
+    before = r._cells[(RequestType.GENERAL, "fast")]
+    await r.record_turn(
+        session_id="fallback",
+        model_name="fast",
+        request_type=RequestType.GENERAL,
+        turn=Turn(user_content="fix this retry bug", assistant_content="clear the cache"),
+    )
+    await r.record_turn(
+        session_id="fallback",
+        model_name="external-default",
+        request_type=RequestType.GENERAL,
+        turn=Turn(user_content="the fix is still broken", assistant_content="keep cache entries"),
+    )
+    assert r._cells[(RequestType.GENERAL, "fast")].beta > before.beta
+    await r.record_turn(
+        session_id="fallback",
+        model_name="smart",
+        request_type=RequestType.GENERAL,
+        turn=Turn(
+            user_content="the fix is still broken",
+            assistant_content="use the corrected entry",
+            tool_results=[{"is_error": True, "content": "failure"}],
+        ),
+    )
+    assert r._feedback_contexts["fallback"].model_name == "smart"
+    assert all(model != "external-default" for _, model in r._cells)
+    assert r.config.available_models == ["fast", "smart"]
+
+
 @pytest.mark.asyncio
 async def test_record_turn_bounds_feedback_contexts_and_evicts_least_recent_session():
     r = _make_router()

tests/test_litellm/router_strategy/test_complexity_router.py

@@ -5,18 +5,19 @@ Tests the rule-based complexity scoring and tier assignment logic.
 """
 
 import asyncio
-from collections.abc import AsyncIterator
 import json
-from copy import deepcopy
-from functools import partial
 import logging
 import sys
 import time
-from typing import Dict, Final, List
+from collections.abc import AsyncIterator, Mapping
+from copy import deepcopy
+from functools import partial
+from typing import Dict, Final, List, Literal
 from unittest.mock import AsyncMock, MagicMock, patch
 
-import pytest
 import httpx
+import pytest
+import respx
 from pydantic import ValidationError
 
 import litellm
@@ -2671,7 +2672,8 @@ class TestEncryptedTaskClassifier:
         assert call["metadata"]["user_api_key_hash"] == "caller-key-hash"
         assert call["proxy_server_request"]["body"]["input"] == call["input"]
         assert call["proxy_server_request"]["originating_request_masked"] == {
-            "input": [task], "metadata": {"authorization": "REDACTED"},
+            "input": [task],
+            "metadata": {"authorization": "REDACTED"},
         }
         assert "source-secret" not in json.dumps(call)
         assert "originating_request_masked" not in call["proxy_server_request"]["body"]
@@ -3295,19 +3297,23 @@ class TestLLMClassifier:
         ]
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("source_body", [
-        {"model": "router", "messages": [{"role": "user", "content": "source-only"}]},
-        {"model": "router", "system": "source-only", "messages": [{"role": "user", "content": "ask"}]},
-        {"model": "router", "instructions": "source-only", "input": "ask"},
-    ])
+    @pytest.mark.parametrize(
+        "source_body",
+        [
+            {"model": "router", "messages": [{"role": "user", "content": "source-only"}]},
+            {"model": "router", "system": "source-only", "messages": [{"role": "user", "content": "ask"}]},
+            {"model": "router", "instructions": "source-only", "input": "ask"},
+        ],
+    )
     async def test_classifier_source_is_masked_and_separate_from_provider_input(
         self, llm_complexity_router, mock_router_instance, source_body
     ):
         mock_router_instance.acompletion = AsyncMock(return_value=_llm_response('{"tier": "SIMPLE"}'))
         outcome = await llm_complexity_router.aclassify(
-            "classify-this-ask", request_kwargs={"proxy_server_request": {
-                "body": {**source_body, "metadata": {"authorization": "source-secret"}}
-            }}
+            "classify-this-ask",
+            request_kwargs={
+                "proxy_server_request": {"body": {**source_body, "metadata": {"authorization": "source-secret"}}}
+            },
         )
         assert outcome.cause == "llm_classifier"
         call_kwargs = mock_router_instance.acompletion.call_args.kwargs
@@ -8011,7 +8017,9 @@ class TestContextAwareClassifier:
             ),
         ),
     )
-    def test_only_text_reminder_tails_are_ignored_for_new_asks(self, tail: list[dict[str, object]], expected: bool) -> None:
+    def test_only_text_reminder_tails_are_ignored_for_new_asks(
+        self, tail: list[dict[str, object]], expected: bool
+    ) -> None:
         from litellm.router_strategy.complexity_router.complexity_router import (
             _CODEX_REMINDER_MARKERS,
             _newest_turn_is_human_ask,
@@ -13095,6 +13103,497 @@ class TestModalityRouting:
         assert cache.async_set_cache.await_args.kwargs["value"] == {"model": "text-cheap", "tier": "SIMPLE"}
 
 
+@pytest.mark.usefixtures("local_model_cost_map")
+class TestHealthFallbackDispatch:
+    @pytest.fixture(autouse=True)
+    def httpx_transport(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+
+    @staticmethod
+    def _router(
+        surface: str = "chat",
+        *,
+        peer: bool = False,
+        session: bool = False,
+        tagged: bool = False,
+        config: Mapping[str, object] | None = None,
+    ) -> Router:
+        provider: Final = "anthropic/claude-sonnet-5" if surface == "messages" else "openai/gpt-5.6"
+        base_suffix: Final = "" if surface == "messages" else "/v1"
+        return Router(
+            model_list=[
+                {
+                    "model_name": "health-router",
+                    "litellm_params": {
+                        "model": "auto_router/complexity_router",
+                        "complexity_router_default_model": (config or {}).get("default_model", "fallback"),
+                        "complexity_router_config": {
+                            "tiers": {"SIMPLE": ["primary", "peer"] if peer else "primary", "MEDIUM": "primary"},
+                            "session_affinity": session,
+                            "deployment_affinity": False,
+                            "max_tokens_from_tier_model": False,
+                            **(config or {}),
+                        },
+                    },
+                },
+                *[
+                    {
+                        "model_name": name,
+                        "litellm_params": {
+                            "model": provider,
+                            "api_key": "test-only",
+                            "api_base": f"https://{name}.test{base_suffix}",
+                            **({"tags": [name]} if tagged else {}),
+                        },
+                        "model_info": {"id": f"{name}-id"},
+                    }
+                    for name in ("primary", "peer", "fallback")
+                ],
+            ],
+            num_retries=0,
+            enable_health_check_routing=True,
+            enable_tag_filtering=tagged,
+        )
+
+    @staticmethod
+    def _unavailable(router: Router, model_id: str, source: Literal["health", "cooldown"]) -> None:
+        if source == "health":
+            router.health_state_cache.set_deployment_health_states(
+                {model_id: {"is_healthy": False, "timestamp": time.time()}}
+            )
+        else:
+            router.cooldown_cache.add_deployment_to_cooldown(
+                model_id=model_id,
+                original_exception=RuntimeError("unavailable"),
+                exception_status=503,
+                cooldown_time=60,
+            )
+
+    @staticmethod
+    def _http_response(request: httpx.Request) -> httpx.Response:
+        body: Final = json.loads(request.content)
+        text: Final = request.url.host.split(".")[0]
+        payload: Final[Mapping[str, object]]
+        events: Final[tuple[Mapping[str, object], ...]]
+        if request.url.path.endswith("/responses"):
+            from litellm.responses.main import mock_responses_api_response
+
+            payload = mock_responses_api_response(text).model_dump()
+            events = (
+                {"type": "response.created", "response": {**payload, "status": "in_progress"}, "sequence_number": 0},
+                {
+                    "type": "response.output_text.delta",
+                    "delta": text,
+                    "item_id": "msg_test",
+                    "output_index": 0,
+                    "content_index": 0,
+                    "sequence_number": 1,
+                },
+                {"type": "response.completed", "response": payload, "sequence_number": 2},
+            )
+        elif request.url.path.endswith("/messages"):
+            payload = {
+                "id": "msg_test",
+                "type": "message",
+                "role": "assistant",
+                "model": body["model"],
+                "content": [{"type": "text", "text": text}],
+                "stop_reason": "end_turn",
+                "stop_sequence": None,
+                "usage": {"input_tokens": 10, "output_tokens": 1},
+            }
+            events = (
+                {"type": "message_start", "message": {**payload, "content": [], "stop_reason": None}},
+                {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+                {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}},
+                {"type": "content_block_stop", "index": 0},
+                {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 1}},
+                {"type": "message_stop"},
+            )
+        else:
+            payload = {
+                "id": "chatcmpl-test",
+                "object": "chat.completion",
+                "created": 1,
+                "model": body["model"],
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": text}, "finish_reason": "stop"}],
+                "usage": {"prompt_tokens": 10, "completion_tokens": 1, "total_tokens": 11},
+            }
+            events = (
+                {
+                    **payload,
+                    "object": "chat.completion.chunk",
+                    "choices": [{"index": 0, "delta": {"content": text}, "finish_reason": None}],
+                },
+                {
+                    **payload,
+                    "object": "chat.completion.chunk",
+                    "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
+                },
+            )
+        if not body.get("stream"):
+            return httpx.Response(200, json=payload)
+        wire: Final = "".join(
+            (f"event: {event['type']}\n" if "type" in event else "") + f"data: {json.dumps(event)}\n\n"
+            for event in events
+        )
+        return httpx.Response(
+            200,
+            text=wire + ("data: [DONE]\n\n" if "type" not in events[0] else ""),
+            headers={"content-type": "text/event-stream"},
+        )
+
+    @staticmethod
+    async def _request(router: Router, surface: str, stream: bool, metadata: dict[str, object]) -> str:
+        if surface == "responses":
+            result = await router.aresponses(
+                model="health-router", input="Hello!", stream=stream, litellm_metadata=metadata
+            )
+        elif surface == "messages":
+            result = await router.aanthropic_messages(
+                model="health-router",
+                messages=[{"role": "user", "content": "Hello!"}],
+                max_tokens=32,
+                stream=stream,
+                litellm_metadata=metadata,
+            )
+        else:
+            result = await router.acompletion(
+                model="health-router",
+                messages=[{"role": "user", "content": "Hello!"}],
+                stream=stream,
+                metadata=metadata,
+            )
+        if not stream:
+            payload = result if isinstance(result, dict) else result.model_dump()
+            if surface == "responses":
+                return payload["output"][0]["content"][0]["text"]
+            if surface == "messages":
+                return payload["content"][0]["text"]
+            return payload["choices"][0]["message"]["content"]
+        if surface == "messages":
+            wire: Final = b"".join([chunk async for chunk in result]).decode()
+            events = tuple(json.loads(line[6:]) for line in wire.splitlines() if line.startswith("data: "))
+            assert events[-1]["type"] == "message_stop"
+            return "".join(c["delta"]["text"] for c in events if c["type"] == "content_block_delta")
+        chunks: Final = [chunk.model_dump() async for chunk in result]
+        if surface == "responses":
+            assert chunks[-1]["type"] == "response.completed"
+            return "".join(c["delta"] for c in chunks if c["type"] == "response.output_text.delta")
+        assert chunks[-1]["choices"][0]["finish_reason"] == "stop"
+        return "".join(c["choices"][0]["delta"].get("content") or "" for c in chunks if c["choices"])
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("surface", ["chat", "responses", "messages"])
+    @pytest.mark.parametrize("stream", [False, True])
+    @pytest.mark.parametrize("source", ["health", "cooldown"])
+    async def test_public_call_falls_back_and_recovers(
+        self, surface: str, stream: bool, source: Literal["health", "cooldown"]
+    ) -> None:
+        router: Final = self._router(surface, session=True)
+        self._unavailable(router, "primary-id", source)
+        metadata: Final[dict[str, object]] = {"session_id": "outage"}
+        with respx.mock(assert_all_mocked=True) as upstream:
+            upstream.post(host__regex=r"^(primary|peer|fallback)\.test$").mock(side_effect=self._http_response)
+            assert await self._request(router, surface, stream, metadata) == "fallback"
+            assert metadata["routing_decision"]["cause"] == "health_default_fallback"
+            assert "tier" not in metadata["routing_decision"]
+            assert "health_displaced:primary" in metadata["routing_decision"]["signals"]
+            assert [c.request.url.host for c in upstream.calls] == ["fallback.test"]
+            strategy: Final = router.complexity_routers["health-router"][0].strategy
+            key: Final = strategy._get_session_affinity_cache_key("outage", {})
+            assert await router.cache.async_get_cache(key=key) is None
+            if source == "health":
+                router.health_state_cache.set_deployment_health_states(
+                    {"primary-id": {"is_healthy": True, "timestamp": time.time()}}
+                )
+            else:
+                router.cooldown_cache.cooldown_store.delete_cache(
+                    router.cooldown_cache.get_cooldown_cache_key("primary-id")
+                )
+            recovered: Final[dict[str, object]] = {"session_id": "outage"}
+            assert await self._request(router, surface, stream, recovered) == "primary"
+            assert recovered["routing_decision"]["routed_model"] == "primary"
+            assert [c.request.url.host for c in upstream.calls] == ["fallback.test", "primary.test"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("source", ["health", "cooldown"])
+    async def test_partial_group_then_peer_then_default(self, source: Literal["health", "cooldown"]) -> None:
+        router: Final = self._router(peer=True, session=True)
+        router.add_deployment(
+            Deployment(
+                model_name="primary",
+                litellm_params=LiteLLM_Params(
+                    model="openai/gpt-5.6", api_key="test-only", api_base="https://primary.test/v1"
+                ),
+                model_info={"id": "primary-sibling-id"},
+            )
+        )
+        strategy: Final = router.complexity_routers["health-router"][0].strategy
+        key: Final = strategy._get_session_affinity_cache_key("precedence", {})
+        await router.cache.async_set_cache(key=key, value={"model": "primary", "tier": "SIMPLE"}, ttl=600)
+        with respx.mock(assert_all_mocked=True) as upstream:
+            upstream.post(host__regex=r"^(primary|peer|fallback)\.test$").mock(side_effect=self._http_response)
+            for model_id, expected, cause in (
+                ("primary-id", "primary", "session_affinity_pin"),
+                ("primary-sibling-id", "peer", "health_failover"),
+                ("peer-id", "fallback", "health_default_fallback"),
+            ):
+                self._unavailable(router, model_id, source)
+                metadata: Final[dict[str, object]] = {"session_id": "precedence"}
+                assert await self._request(router, "chat", False, metadata) == expected
+                assert metadata["routing_decision"]["cause"] == cause
+                assert await router.cache.async_get_cache(key=key) == {"model": "primary", "tier": "SIMPLE"}
+            assert [c.request.url.host for c in upstream.calls] == ["primary.test", "peer.test", "fallback.test"]
+
+    @pytest.mark.asyncio
+    async def test_concurrent_tag_scopes_keep_fallbacks_request_local(self) -> None:
+        router: Final = self._router(tagged=True)
+        router.add_deployment(
+            Deployment(
+                model_name="fallback",
+                litellm_params=LiteLLM_Params(
+                    model="openai/gpt-5.6", api_key="test-only", api_base="https://peer.test/v1", tags=["peer"]
+                ),
+                model_info={"id": "fallback-peer-id"},
+            )
+        )
+        self._unavailable(router, "primary-id", "cooldown")
+        with respx.mock(assert_all_mocked=True) as upstream:
+            upstream.post(host__regex=r"^(peer|fallback)\.test$").mock(side_effect=self._http_response)
+            scopes: Final = tuple({"tags": [name], "session_id": name} for name in ("peer", "fallback"))
+            results: Final = await asyncio.gather(
+                *(self._request(router, "chat", False, metadata) for metadata in scopes)
+            )
+            assert results == ["peer", "fallback"]
+            assert [m["tags"] for m in scopes] == [["peer"], ["fallback"]]
+            assert [m["routing_decision"]["routed_model"] for m in scopes] == ["fallback", "fallback"]
+            assert sorted(c.request.url.host for c in upstream.calls) == ["fallback.test", "peer.test"]
+
+    @pytest.mark.asyncio
+    async def test_probe_preserves_consumed_request_exclusions(self) -> None:
+        router: Final = self._router()
+        self._unavailable(router, "primary-id", "cooldown")
+        kwargs: Final = {"_excluded_deployment_ids": ["fallback-id"], "_target_order": 1}
+        strategy: Final = router.complexity_routers["health-router"][0].strategy
+        response: Final = await strategy.async_pre_routing_hook(
+            model="health-router", messages=[{"role": "user", "content": "Hello!"}], request_kwargs=kwargs
+        )
+        assert response.model == "primary"
+        assert kwargs == {"_excluded_deployment_ids": ["fallback-id"], "_target_order": 1}
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("default_state", ["cooldown", "unconfigured", "same-model"])
+    async def test_unavailable_default_preserves_no_deployment_error(self, default_state: str) -> None:
+        from litellm.types.router import RouterRateLimitError
+
+        router: Final = self._router(config={"default_model": "primary"} if default_state == "same-model" else None)
+        self._unavailable(router, "primary-id", "cooldown")
+        if default_state == "unconfigured":
+            router.delete_deployment(id="fallback-id")
+        elif default_state == "cooldown":
+            self._unavailable(router, "fallback-id", "cooldown")
+        with respx.mock(assert_all_mocked=True) as upstream:
+            with pytest.raises(RouterRateLimitError, match="No deployments available"):
+                await self._request(router, "chat", False, {})
+            assert not upstream.calls
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("plan_active", [False, True])
+    async def test_plan_floor_outage_cannot_use_untiered_default(self, plan_active: bool) -> None:
+        from litellm.types.router import RouterRateLimitError
+
+        router: Final = self._router(
+            config={"tiers": {"SIMPLE": "primary", "MEDIUM": "peer"}, "plan_mode_min_tier": "MEDIUM"}
+        )
+        self._unavailable(router, "primary-id", "cooldown")
+        self._unavailable(router, "peer-id", "cooldown")
+        metadata: Final = {}
+        with respx.mock(assert_all_mocked=True, assert_all_called=False) as upstream:
+            upstream.post(host="fallback.test").mock(side_effect=self._http_response)
+            if plan_active:
+                with pytest.raises(RouterRateLimitError, match="No deployments available"):
+                    await router.acompletion(
+                        model="health-router",
+                        messages=[
+                            {"role": "system", "content": "Plan mode is active"},
+                            {"role": "user", "content": "Hello!"},
+                        ],
+                        metadata=metadata,
+                    )
+                assert not upstream.calls
+                assert metadata["routing_decision"]["routed_model"] == "peer"
+                assert metadata["routing_decision"]["tier"] == "MEDIUM"
+            else:
+                assert await self._request(router, "chat", False, metadata) == "fallback"
+
+    @pytest.mark.asyncio
+    async def test_default_dispatch_drops_displaced_tier_params(self) -> None:
+        router: Final = self._router(
+            config={"tiers": {"SIMPLE": {"model_name": "primary", "litellm_params": {"max_tokens": 9}}}}
+        )
+        with respx.mock(assert_all_mocked=True) as upstream:
+            upstream.post(host__regex=r"^(primary|fallback)\.test$").mock(side_effect=self._http_response)
+            await router.acompletion(
+                model="health-router", messages=[{"role": "user", "content": "Hello!"}], max_tokens=32
+            )
+            assert json.loads(upstream.calls[-1].request.content)["max_completion_tokens"] == 9
+            self._unavailable(router, "primary-id", "cooldown")
+            await router.acompletion(
+                model="health-router", messages=[{"role": "user", "content": "Hello!"}], max_tokens=32
+            )
+            assert json.loads(upstream.calls[-1].request.content)["max_completion_tokens"] == 32
+            assert upstream.calls[-1].request.url.host == "fallback.test"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("source", ["health", "cooldown"])
+    async def test_pinned_session_returns_to_primary_after_outage(self, source: Literal["health", "cooldown"]) -> None:
+        router: Final = self._router(session=True)
+        with respx.mock(assert_all_mocked=True) as upstream:
+            upstream.post(host__regex=r"^(primary|fallback)\.test$").mock(side_effect=self._http_response)
+            assert await self._request(router, "chat", False, {"session_id": "pinned"}) == "primary"
+            self._unavailable(router, "primary-id", source)
+            outage: Final[dict[str, object]] = {"session_id": "pinned"}
+            assert await self._request(router, "chat", False, outage) == "fallback"
+            assert outage["routing_decision"]["cause"] == "health_default_fallback"
+            if source == "health":
+                router.health_state_cache.set_deployment_health_states(
+                    {"primary-id": {"is_healthy": True, "timestamp": time.time()}}
+                )
+            else:
+                router.cooldown_cache.cooldown_store.delete_cache(
+                    router.cooldown_cache.get_cooldown_cache_key("primary-id")
+                )
+            recovered: Final[dict[str, object]] = {"session_id": "pinned"}
+            assert await self._request(router, "chat", False, recovered) == "primary"
+            assert recovered["routing_decision"]["cause"] == "session_affinity_pin"
+            assert [c.request.url.host for c in upstream.calls] == ["primary.test", "fallback.test", "primary.test"]
+
+    @pytest.mark.asyncio
+    async def test_policy_plugin_does_not_escape_to_live_default(self) -> None:
+        from litellm.types.router import RouterRateLimitError, RoutingContext
+
+        class PrimaryOnly:
+            async def run(self, context: RoutingContext) -> RoutingContext:
+                context.candidate_models = [name for name in context.candidate_models if name == "primary"]
+                return context
+
+        router: Final = self._router(peer=True, config={"plugins": [PrimaryOnly()]})
+        self._unavailable(router, "primary-id", "cooldown")
+        with respx.mock(assert_all_mocked=True) as upstream:
+            with pytest.raises(RouterRateLimitError, match="No deployments available"):
+                await self._request(router, "chat", False, {})
+            assert not upstream.calls
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("live_tier", [True, False])
+    @pytest.mark.parametrize("default_fits", [True, False])
+    async def test_context_recovery_precedes_default_with_prechecks_off(
+        self, live_tier: bool, default_fits: bool
+    ) -> None:
+        from litellm.types.router import RouterRateLimitError
+
+        router: Final = self._router(config={"tiers": {"SIMPLE": "primary", "MEDIUM": "peer", "COMPLEX": "large"}})
+        router.add_deployment(
+            Deployment(
+                model_name="large",
+                litellm_params=LiteLLM_Params(
+                    model="openai/gpt-5.6", api_key="test-only", api_base="https://large.test/v1"
+                ),
+                model_info={"id": "large-id", "max_input_tokens": 10000},
+            )
+        )
+        for deployment in router.model_list:
+            deployment["model_info"]["max_input_tokens"] = (
+                10
+                if deployment["model_name"] == "primary"
+                or (deployment["model_name"] == "fallback" and not default_fits)
+                else 10000
+            )
+        self._unavailable(router, "peer-id", "cooldown")
+        if not live_tier:
+            self._unavailable(router, "large-id", "cooldown")
+        assert router.enable_pre_call_checks is False
+        metadata: Final = {}
+        messages: Final = [{"role": "user", "content": "hello " * 100}]
+        with respx.mock(assert_all_mocked=True, assert_all_called=False) as upstream:
+            upstream.post(host__regex=r"^(large|fallback)\.test$").mock(side_effect=self._http_response)
+            if not live_tier and not default_fits:
+                with pytest.raises(RouterRateLimitError, match="No deployments available"):
+                    await router.acompletion(model="health-router", messages=messages, metadata=metadata)
+                assert not upstream.calls
+            else:
+                result: Final = await router.acompletion(model="health-router", messages=messages, metadata=metadata)
+                expected: Final = "large" if live_tier else "fallback"
+                assert result.choices[0].message.content == expected
+                assert upstream.calls[-1].request.url.host == f"{expected}.test"
+                assert metadata["routing_decision"].get("tier") == ("COMPLEX" if live_tier else None)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("live_tier", [True, False])
+    async def test_modality_recovery_precedes_default(self, live_tier: bool) -> None:
+        router: Final = self._router(
+            config={"modality_routing": True, "tiers": {"SIMPLE": "primary", "MEDIUM": "peer", "COMPLEX": "vision"}}
+        )
+        router.add_deployment(
+            Deployment(
+                model_name="vision",
+                litellm_params=LiteLLM_Params(
+                    model="openai/gpt-5.6", api_key="test-only", api_base="https://vision.test/v1"
+                ),
+                model_info={"id": "vision-id", "supports_vision": True},
+            )
+        )
+        for deployment in router.model_list:
+            deployment["model_info"]["supports_vision"] = deployment["model_name"] != "primary"
+        self._unavailable(router, "peer-id", "cooldown")
+        if not live_tier:
+            self._unavailable(router, "vision-id", "cooldown")
+        with respx.mock(assert_all_mocked=True) as upstream:
+            upstream.post(host__regex=r"^(vision|fallback)\.test$").mock(side_effect=self._http_response)
+            result: Final = await router.acompletion(
+                model="health-router",
+                messages=[
+                    {
+                        "role": "user",
+                        "content": [
+                            {"type": "text", "text": "Hello!"},
+                            {"type": "image_url", "image_url": {"url": "data:image/png;base64,aGk="}},
+                        ],
+                    }
+                ],
+            )
+            expected: Final = "vision" if live_tier else "fallback"
+            assert result.choices[0].message.content == expected
+            assert upstream.calls[-1].request.url.host == f"{expected}.test"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("default_fits", [True, False])
+    async def test_modality_default_must_also_fit_context(self, default_fits: bool) -> None:
+        router: Final = self._router(config={"modality_routing": True, "tiers": {"SIMPLE": "primary"}})
+        for deployment in router.model_list:
+            deployment["model_info"]["supports_vision"] = deployment["model_name"] == "fallback"
+            deployment["model_info"]["max_input_tokens"] = 10000 if default_fits else 10
+        with respx.mock(assert_all_mocked=True, assert_all_called=False) as upstream:
+            upstream.post(host="fallback.test").mock(side_effect=self._http_response)
+            messages: Final = [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": "hello " * 100},
+                        {"type": "image_url", "image_url": {"url": "data:image/png;base64,aGk="}},
+                    ],
+                }
+            ]
+            if default_fits:
+                result: Final = await router.acompletion(model="health-router", messages=messages)
+                assert result.choices[0].message.content == "fallback"
+            else:
+                with pytest.raises(litellm.BadRequestError, match="modality_routing is enabled"):
+                    await router.acompletion(model="health-router", messages=messages)
+                assert not upstream.calls
+
+
 class TestTierHealthFailover:
     """A tier whose decided model group is entirely in cooldown falls back to a live peer."""
 
@@ -13129,7 +13628,7 @@ class TestTierHealthFailover:
         probed_prompts = []
 
         async def get_healthy_deployments(
-            model, request_kwargs, messages=None, input=None, parent_otel_span=None, **kwargs
+            model, request_kwargs, messages=None, input=None, parent_otel_span=None, health_check_probe=False
         ):
             probed_kwargs.append(request_kwargs)
             probed_prompts.append((messages, input))

tests/test_litellm/test_router.py

@@ -7381,6 +7381,63 @@ async def test_async_get_fully_unhealthy_model_names_marks_name_when_all_unhealt
     assert await router.async_get_fully_unhealthy_model_names() == {"gpt-4o"}
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("health_check_probe", [False, True])
+@pytest.mark.parametrize(
+    "state, health_routing, fails_policy, scoped, strict_ids",
+    [
+        ("absent", True, False, False, ("dep-0", "dep-1")),
+        ("partial", True, False, False, ("dep-1",)),
+        ("all", True, False, False, ()),
+        ("stale", True, False, False, ("dep-0", "dep-1")),
+        ("all", False, False, False, ("dep-0", "dep-1")),
+        ("all", True, True, False, ("dep-0", "dep-1")),
+        ("all", True, True, True, ()),
+    ],
+)
+async def test_health_probe_preserves_normal_caller_policy(
+    health_check_probe: bool,
+    state: str,
+    health_routing: bool,
+    fails_policy: bool,
+    scoped: bool,
+    strict_ids: tuple[str, ...],
+) -> None:
+    import time
+    from litellm.types.router import AllowedFailsPolicy, RouterRateLimitError
+
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": "health-group",
+                "litellm_params": {"model": "openai/gpt-5.6", "api_key": "test-only"},
+                "model_info": {"id": model_id},
+            }
+            for model_id in ("dep-0", "dep-1")
+        ],
+        enable_health_check_routing=health_routing,
+        allowed_fails_policy=AllowedFailsPolicy(ServiceUnavailableErrorAllowedFails=2) if fails_policy else None,
+        background_health_check_model_groups=["health-group"] if scoped else None,
+    )
+    if state != "absent":
+        _seed_unhealthy_states(
+            router,
+            ("dep-0",) if state == "partial" else ("dep-0", "dep-1"),
+            time.time() - router.health_state_cache.staleness_threshold - 10 if state == "stale" else None,
+        )
+    expected: Final = strict_ids if strict_ids or health_check_probe else ("dep-0", "dep-1")
+    if not expected:
+        with pytest.raises(RouterRateLimitError, match="No deployments available"):
+            await router.async_get_healthy_deployments(model="health-group", request_kwargs={}, health_check_probe=True)
+    else:
+        deployments: Final = await router.async_get_healthy_deployments(
+            model="health-group", request_kwargs={}, health_check_probe=health_check_probe
+        )
+        assert {d["model_info"]["id"] for d in deployments} == set(expected)
+    assert await router.cooldown_cache.async_get_active_cooldowns(["dep-0", "dep-1"], parent_otel_span=None) == []
+
+
+
 @pytest.mark.asyncio
 async def test_async_get_fully_unhealthy_model_names_keeps_name_when_partial():
     router = _router_with_two_deployments([False, False])

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -36460,7 +36460,7 @@ export interface components {
              * Cause
              * @enum {string}
              */
-            cause?: "heuristic_scorer" | "heuristic_v2" | "reasoning_override" | "llm_classifier" | "heuristic_first_short_circuit" | "hybrid_short_circuit" | "classifier_plugin" | "classifier_fallback" | "default_model_fallback" | "literal_keyword_match" | "semantic_keyword_match" | "plan_mode" | "housekeeping" | "modality_escalation" | "modality_pin_override" | "health_failover" | "session_affinity_pin" | "session_affinity_escalation" | "user_turn_continuation" | "default_fallback" | "keyword" | "quality_tier" | "bandit";
+            cause?: "heuristic_scorer" | "heuristic_v2" | "reasoning_override" | "llm_classifier" | "heuristic_first_short_circuit" | "hybrid_short_circuit" | "classifier_plugin" | "classifier_fallback" | "default_model_fallback" | "literal_keyword_match" | "semantic_keyword_match" | "plan_mode" | "housekeeping" | "modality_escalation" | "modality_pin_override" | "health_failover" | "health_default_fallback" | "session_affinity_pin" | "session_affinity_escalation" | "user_turn_continuation" | "default_fallback" | "keyword" | "quality_tier" | "bandit";
             /** Classifier Cost */
             classifier_cost?: number;
             /** Classifier Model */