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

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

Co-Authored-By: Claude Code <noreply@anthropic.com>
(cherry picked from commit 00c7fd8376)

* fix(router): treat budget and tag exhaustion as a no-capacity verdict

The eligibility probe only read typed router errors as "nothing here can
serve this". Provider and deployment budget exhaustion, and tag routing
with no matching deployment, report it as a bare ValueError carrying a
RouterErrors marker, so the probe read a spent tier as live, skipped the
peer and default recovery, and failed the request.

---------

Co-authored-by: Tin Chi Lo <tin@berri.ai>
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

@@ -937,6 +937,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 ())
@@ -1098,6 +1099,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)."""
@@ -2681,12 +2691,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
@@ -2698,17 +2728,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,
@@ -2720,14 +2743,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)
@@ -2735,7 +2760,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(
@@ -2881,6 +2906,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.
 
@@ -2911,7 +2937,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
@@ -3030,12 +3057,15 @@ class ComplexityRouter(CustomLogger):
 
         Every way the owner says "nothing here can serve this" is a negative verdict: no healthy
         deployment for the group at all (BadRequestError, which ContextWindowExceededError
-        subclasses), every deployment filtered out (RouterRateLimitError), and every deployment
-        over its RPM (RouterRateLimitErrorBasic). Anything else is unknown rather than negative,
-        so it reads as capacity: absent information must never decide the verdict.
+        subclasses), every deployment filtered out (RouterRateLimitError), every deployment over
+        its RPM (RouterRateLimitErrorBasic), and every deployment refused by a filter that reports
+        exhaustion as a bare ValueError naming a RouterErrors marker -- provider and deployment
+        budgets, and tag routing, which have no typed error of their own. Anything else is unknown
+        rather than negative, so it reads as capacity: absent information must never decide the
+        verdict.
         """
         from litellm.exceptions import BadRequestError
-        from litellm.types.router import RouterRateLimitError, RouterRateLimitErrorBasic
+        from litellm.types.router import RouterErrors, RouterRateLimitError, RouterRateLimitErrorBasic
 
         probe_kwargs: Final = dict(request_kwargs)  # mutable-ok: the owner pops routing keys off the dict it is handed
         try:
@@ -3045,10 +3075,15 @@ 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
+            if isinstance(exc, ValueError) and any(marker.value in str(exc) for marker in RouterErrors):
+                verbose_router_logger.debug("health probe exhausted model=%s error=%s", model_name, exc)
+                return False
             verbose_router_logger.debug(
                 "ComplexityRouter: eligibility probe for %s failed, treating the group as live: %s", model_name, exc
             )
@@ -3062,76 +3097,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"),
@@ -3140,14 +3223,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,
             }
         )
 
@@ -3462,6 +3545,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)
 
@@ -3512,7 +3596,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
@@ -3582,11 +3670,13 @@ class ComplexityRouter(CustomLogger):
                                 messages,
                                 resolved_messages,
                                 request_kwargs,
+                                context_fit,
                             ),
                             messages,
                             input,
                             resolved_messages,
                             request_kwargs,
+                            context_fit,
                         )
                     )
 
@@ -3598,14 +3688,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
@@ -3640,6 +3734,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.
@@ -3811,7 +3906,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
@@ -69,6 +70,7 @@ from litellm.router_strategy.complexity_router.tier_predictor import (
 from litellm.types.router import (
     Deployment,
     LiteLLM_Params,
+    RouterErrors,
     TaggedPreRoutingStrategy,
 )
 from litellm.types.llms.openai import ResponsesAPIResponse
@@ -2671,7 +2673,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"]
@@ -3315,19 +3318,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
@@ -8138,7 +8145,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,
@@ -13222,6 +13231,528 @@ 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,
+        budgeted: 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 {}),
+                            **(
+                                {"max_budget": 1.0, "budget_duration": "1d"}
+                                if budgeted and name == "primary"
+                                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_spent_deployment_budget_falls_back_to_the_default(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """A spent budget leaves the tier with nothing that may serve the request, and the budget
+        filter reports that as a bare ValueError instead of a typed router error. Reading it as
+        capacity skips the recovery and fails the request the recovery exists for."""
+
+        async def _no_sync(*args: object, **kwargs: object) -> None:
+            return None
+
+        monkeypatch.setattr(
+            "litellm.router_strategy.budget_limiter.RouterBudgetLimiting.periodic_sync_in_memory_spend_with_redis",
+            _no_sync,
+        )
+        monkeypatch.setattr(litellm, "callbacks", [])
+        router: Final = self._router(budgeted=True)
+        limiter: Final = router.router_budget_logger
+        assert limiter is not None, "a deployment max_budget must install the budget limiter"
+        await router.cache.async_set_cache(key="deployment_spend:primary-id:1d", value=2.0)
+        with respx.mock(assert_all_mocked=True) as upstream:
+            upstream.post(host__regex=r"^(primary|fallback)\.test$").mock(side_effect=self._http_response)
+            metadata: Final[dict[str, object]] = {}
+            assert await self._request(router, "chat", False, metadata) == "fallback"
+            assert metadata["routing_decision"]["cause"] == "health_default_fallback"
+            assert [c.request.url.host for c in upstream.calls] == ["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."""
 
@@ -13256,7 +13787,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))
@@ -13758,6 +14289,51 @@ class TestTierHealthFailover:
             for _, probed_input in router.litellm_router_instance.probed_prompts
         ), "the eligibility probe must forward `input` to the owner"
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "raised, expected",
+        [
+            (ValueError(f"{RouterErrors.no_deployments_with_tag_routing.value}. Passed model=b"), {"live-c"}),
+            (
+                ValueError(f"{RouterErrors.no_deployments_with_provider_budget_routing.value}: b over budget"),
+                {"live-c"},
+            ),
+            (ValueError("cannot unpack non-sequence"), {"exhausted-b", "live-c"}),
+        ],
+    )
+    async def test_a_marked_exhaustion_value_error_is_a_verdict_and_an_unmarked_one_is_not(
+        self, mock_router_instance, raised, expected
+    ):
+        """Budget and tag filters exhaust a group without a typed error, signalling it only by a
+        RouterErrors marker on a bare ValueError. Those are verdicts; any other ValueError is a
+        fault, and a fault must still read as capacity rather than silently rerouting."""
+        router = self._router(
+            mock_router_instance,
+            {
+                "tiers": {
+                    "SIMPLE": ["dead-a", "exhausted-b", "live-c"],
+                    "MEDIUM": "mid",
+                    "COMPLEX": "big",
+                    "REASONING": "top",
+                },
+                "session_affinity": True,
+            },
+            {"dead-a": ["id-a1"], "exhausted-b": ["id-b1"], "live-c": ["id-c1"]},
+            cooling=("id-a1",),
+            raises_for={"exhausted-b": raised},
+        )
+        key = router._get_session_affinity_cache_key("sess-exhausted", {})
+        await router.litellm_router_instance.cache.async_set_cache(
+            key=key, value={"model": "dead-a", "tier": "SIMPLE"}, ttl=600
+        )
+        results = [
+            await router.async_pre_routing_hook(
+                model="m", request_kwargs={"metadata": {"session_id": "sess-exhausted"}}, messages=self.SIMPLE_MESSAGE
+            )
+            for _ in range(20)
+        ]
+        assert {r.model for r in results} == expected
+
     @pytest.mark.asyncio
     async def test_a_group_the_router_has_no_deployment_for_is_not_a_failover_target(self, mock_router_instance):
         """The owner answers an unconfigured group with BadRequestError. Reading that as live

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

@@ -36329,7 +36329,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 */