fix(complexity_router): resolve a classified tier up a ladder instead of to whatever is configured

A tier resolved to whatever the config named, whether or not the router could serve it.
A model group that was never registered, or whose every deployment was in cooldown,
still won the pick and failed the request. A tier with no entry at all jumped to
default_model and then to MEDIUM, which could answer a COMPLEX request with the model
the classifier had just ruled out

Resolution now walks a ladder: live models in the classified tier, then each tier above
it, then default_model, then the classified tier as a best effort. A pool member counts
as live when the router knows that model group and it has at least one deployment out
of cooldown, so a dead model is passed over for its peers before the tier is given up
on. Health probes for a pool run concurrently and the ladder stops at the first tier
that can serve, so the common case is one tier's worth of cooldown reads

It never falls to a cheaper tier. That is the model the classifier already ruled out,
so a COMPLEX request is not answered by the SIMPLE model just because SIMPLE is
healthy. The last step is best effort rather than an error because cooldowns expire and
the health view is a snapshot; a request that might succeed is sent rather than failed

Climbing is recorded as tier_fallback_from on the routing decision and in the log line,
so a cheap request sitting on an expensive model is explainable afterwards. `tier` is
the tier that served, matching how the escalation path already reports the final tier

Two behaviors are deliberately kept. A tier configured as an empty pool still raises,
being a config error rather than a gap, and climbing past it would hide the
misconfiguration behind a pricier model. A routing plugin narrowing a tier to zero still
raises, because refusing every candidate is a policy decision; plugins now run against
whichever tier the ladder settles on, so they vet what is actually served rather than
being handed a tier that cannot answer. An unreadable health view treats the pool as
live, degrading to the previous behavior instead of declaring every tier dead

litellm/router_strategy/complexity_router/README.md

@@ -126,6 +126,25 @@ response = litellm.completion(
 
 ## Special Behaviors
 
+### Tier Fallback Ladder
+
+Classifying a request is only half the job; the tier still has to have something that can answer. Resolution runs in this order:
+
+1. A live model in the classified tier. A pool member counts as live when the router knows that model group and it has at least one deployment out of cooldown, so a dead model in a pool is passed over for its peers.
+2. The next tier up, then the one above that. Resolution never falls to a cheaper tier, because that is the model the classifier already ruled out; a request classified COMPLEX is not answered by the SIMPLE model just because SIMPLE is healthy.
+3. `default_model`, if you set one.
+4. The classified tier anyway, as a best effort. Cooldowns expire and the health view is a snapshot, so a request that might succeed is sent rather than failed.
+
+When the ladder climbs, the decision records `tier_fallback_from` with the tier the request was classified into, and the log line carries the same fact:
+
+```
+ComplexityRouter: routing decision cause=heuristic_scorer, tier=MEDIUM, score=-0.150, signals=[...], routed_model=gpt-4o, tier_fallback_from=SIMPLE
+```
+
+Two things are deliberately not routed around. A tier configured as an empty pool (`SIMPLE: []`) raises, because it is a config error rather than a gap. And a routing plugin that narrows a tier to zero candidates raises, because refusing every candidate is a policy decision; climbing past it would serve exactly what the plugin just denied. Plugins run against whichever tier the ladder settles on, so they still vet what is served.
+
+If the health view itself cannot be read, the whole pool is treated as live and resolution behaves as it did before, instead of declaring every tier dead and pushing traffic to the top.
+
 ### Reasoning Override
 
 If 2+ reasoning markers are detected in the user message, the request is automatically routed to the REASONING tier regardless of the weighted score. This ensures complex reasoning tasks get the appropriate model.

litellm/router_strategy/complexity_router/complexity_router.py

@@ -596,6 +596,7 @@ class ComplexityRouter(CustomLogger):
         escalation_keyword: str | None = None,
         escalated: bool = False,
         classifier_model: str | None = None,
+        tier_fallback_from: ComplexityTier | None = None,
     ) -> StandardLoggingRoutingDecision:
         """Assemble the per-request provenance record for this router's decision.
 
@@ -632,6 +633,11 @@ class ComplexityRouter(CustomLogger):
             decision["escalated"] = escalated
         if classifier_model is not None:
             decision["classifier_model"] = classifier_model
+        if tier_fallback_from is not None:
+            # `tier` is what served the request. When the ladder had to climb, the tier the
+            # request was actually classified into is the fact that explains why a cheap
+            # request is on an expensive model, so it rides along rather than being lost.
+            decision["tier_fallback_from"] = tier_fallback_from.value
         return decision
 
     async def aclassify(
@@ -812,19 +818,40 @@ class ComplexityRouter(CustomLogger):
         Returns:
             The model name configured for that tier.
         """
-        tier_key = tier.value if isinstance(tier, ComplexityTier) else tier
+        model, _served_tier = self._resolve_configured_tier(tier)
+        return model
 
-        if tier_key in self.config.tiers:
-            return self._pick_from_tier_value(self.config.tiers[tier_key], tier_key)
+    def _resolve_configured_tier(self, tier: ComplexityTier) -> tuple[str, ComplexityTier]:
+        """Structural half of tier resolution: the first tier at or above `tier` that has
+        models configured, then `default_model` as the last resort.
+
+        Falling upward rather than to a fixed tier keeps an under-configured deployment from
+        quietly serving a request from a cheaper model than it classified.
+        """
+        for candidate_tier in self._tier_ladder_from(tier):
+            if candidate_tier.value not in self.config.tiers:
+                continue
+            # A tier configured as an empty pool is a config error, not a gap to route
+            # around, so _pick_from_tier_value raises rather than climbing past it.
+            return (
+                self._pick_from_tier_value(self.config.tiers[candidate_tier.value], candidate_tier.value),
+                candidate_tier,
+            )
 
         if self.config.default_model:
-            return self.config.default_model
+            return self.config.default_model, tier
 
-        medium_key = ComplexityTier.MEDIUM.value
-        if medium_key in self.config.tiers:
-            return self._pick_from_tier_value(self.config.tiers[medium_key], medium_key)
+        raise ValueError(
+            f"No model configured for tier {tier.value} or any tier above it "
+            f"({sorted(self.config.tiers)}), and no default_model set"
+        )
 
-        raise ValueError(f"No model configured for tier {tier_key} and no default_model set")
+    @staticmethod
+    def _tier_ladder_from(tier: ComplexityTier) -> tuple[ComplexityTier, ...]:
+        """`tier` and every tier above it, cheapest first."""
+        if tier not in TIER_SEVERITY_ORDER:
+            return (tier,)
+        return TIER_SEVERITY_ORDER[TIER_SEVERITY_ORDER.index(tier) :]
 
     @staticmethod
     def _pick_from_tier_value(model: str | list[str], tier_key: str) -> str:
@@ -837,37 +864,103 @@ class ComplexityRouter(CustomLogger):
     def _tier_pools(self) -> dict[str, list[str]]:
         return {tier: (models if isinstance(models, list) else [models]) for tier, models in self.config.tiers.items()}
 
+    async def _has_live_deployment(self, model: str) -> bool:
+        healthy, _ = await self.litellm_router_instance._async_get_healthy_deployments(
+            model=model, parent_otel_span=None
+        )
+        return bool(healthy)
+
+    async def _live_models(self, pool: Sequence[str]) -> tuple[str, ...]:
+        """Pool members the router can serve right now: the model group exists and has at
+        least one deployment that is not in cooldown.
+
+        The health view is advisory. If it cannot be read the whole pool is returned, so a
+        cache or router hiccup degrades to the previous behavior instead of emptying a tier
+        and pushing every request up the ladder.
+        """
+        if not pool:
+            return ()
+        try:
+            live = await asyncio.gather(*(self._has_live_deployment(model) for model in pool))
+            return tuple(model for model, is_live in zip(pool, live) if is_live)
+        except Exception as e:  # noqa: BLE001 -- health lookup can fail in many ways (cache, redis, an unregistered model group); none of them should block routing
+            verbose_router_logger.debug(
+                f"ComplexityRouter: health view unavailable ({e}), treating the tier pool as servable"
+            )
+            return tuple(pool)
+
     async def _pick_model_for_tier(
         self,
         tier: ComplexityTier,
         raw_messages: list[dict[str, Any]] | None,
         resolved_messages: list[dict[str, Any]] | None,
         request_kwargs: dict,
-    ) -> str:
-        if not self.config.plugins:
-            return self.get_model_for_tier(tier)
+    ) -> tuple[str, ComplexityTier]:
+        """Resolve a classified tier to a servable model, returning the tier it came from.
 
+        Walks the tier ladder: the classified tier's live models first, then the next tier
+        up, and so on. `default_model` stays the last resort, and is skipped entirely when
+        routing plugins are configured so their policy cannot be bypassed.
+        """
+        pools = self._tier_pools()
+        for candidate_tier in self._tier_ladder_from(tier):
+            if candidate_tier.value not in pools:
+                continue
+            pool = pools[candidate_tier.value]
+            if not pool:
+                raise ValueError(f"Empty model pool for tier {candidate_tier.value}")
+            live = await self._live_models(pool)
+            if not live:
+                continue
+            if not self.config.plugins:
+                return random.choice(live), candidate_tier
+            narrowed = await self._run_routing_plugins(
+                live, candidate_tier, raw_messages, resolved_messages, request_kwargs
+            )
+            return random.choice(narrowed), candidate_tier
+
+        if self.config.plugins:
+            # default_model was never checked against the plugins, so serving it here would
+            # let their policy be bypassed. Raise instead, matching the Router-level plugin
+            # pipeline's own fail-closed behavior for the same situation.
+            raise ValueError(
+                f"No servable models for tier {tier.value} or any tier above it, with routing plugins configured"
+            )
+        if self.config.default_model:
+            return self.config.default_model, tier
+        # Nothing on the ladder reported a live deployment and there is no declared
+        # fallback. Cooldowns expire and the health view is a snapshot, so serve the
+        # classified tier anyway and let the router surface the real error, rather than
+        # failing a request that may well succeed.
+        return self._resolve_configured_tier(tier)
+
+    async def _run_routing_plugins(
+        self,
+        candidates: Sequence[str],
+        tier: ComplexityTier,
+        raw_messages: Sequence[dict[str, Any]] | None,
+        resolved_messages: Sequence[dict[str, Any]] | None,
+        request_kwargs: dict,
+    ) -> tuple[str, ...]:
         from litellm.types.router import RoutingContext
 
-        tier_key = tier.value
         metadata_key = "litellm_metadata" if "litellm_metadata" in request_kwargs else "metadata"
         context = RoutingContext(
             raw_messages=raw_messages or [],
             structured_messages=resolved_messages or [],
-            candidate_models=list(self._tier_pools().get(tier_key, [])),
+            candidate_models=list(candidates),
             metadata=request_kwargs.get(metadata_key) or {},
         )
-        for plugin in self.config.plugins:
+        for plugin in self.config.plugins or []:
             context = await plugin.run(context)
 
         if not context.candidate_models:
             # A plugin narrowing a tier to zero candidates is a policy decision (e.g. no
-            # model this tenant's budget allows) -- falling back to default_model here
-            # (which was never checked against the plugins) would let that policy be
-            # silently bypassed. Raise instead, matching the Router-level plugin
-            # pipeline's own fail-closed behavior for the same situation.
-            raise ValueError(f"No candidate models left for tier {tier_key} after routing-plugin filtering")
-        return self._pick_from_tier_value(context.candidate_models, tier_key)
+            # model this tenant's budget allows), not a gap to route around. Raise rather
+            # than climbing to the next tier, which would serve exactly what the policy
+            # just refused.
+            raise ValueError(f"No candidate models left for tier {tier.value} after routing-plugin filtering")
+        return tuple(context.candidate_models)
 
     def _ensure_adaptive_router(self) -> Any | None:
         if not self.config.adaptive:
@@ -1437,7 +1530,7 @@ class ComplexityRouter(CustomLogger):
                 # Plugins configured: default_model must never bypass them, so it's not
                 # checked here at all -- _pick_model_for_tier -> get_model_for_tier still
                 # falls back to it (after the MEDIUM tier) once the plugin pipeline runs.
-                routed_model = await self._pick_model_for_tier(
+                routed_model, _ = await self._pick_model_for_tier(
                     ComplexityTier.MEDIUM, messages, resolved_messages, request_kwargs
                 )
             return PreRoutingHookResponse(
@@ -1453,7 +1546,9 @@ class ComplexityRouter(CustomLogger):
         if override is not None:
             routed_tier = self._escalate_tier(override.tier) if escalation_keyword is not None else override.tier
             keyword_escalated = routed_tier != override.tier
-            routed_model = await self._pick_model_for_tier(routed_tier, messages, resolved_messages, request_kwargs)
+            routed_model, keyword_served_tier = await self._pick_model_for_tier(
+                routed_tier, messages, resolved_messages, request_kwargs
+            )
             keyword_cause: RoutingDecisionCause = (
                 "semantic_keyword_match" if self.config.semantic_keyword_matching else "literal_keyword_match"
             )
@@ -1467,10 +1562,11 @@ class ComplexityRouter(CustomLogger):
                 routing_decision=self._build_routing_decision(
                     routed_model=routed_model,
                     cause=keyword_cause,
-                    tier=routed_tier,
+                    tier=keyword_served_tier,
                     matched_keyword=override.matched_keyword,
                     escalation_keyword=escalation_keyword,
                     escalated=keyword_escalated,
+                    tier_fallback_from=routed_tier if keyword_served_tier != routed_tier else None,
                 ),
             )
 
@@ -1483,6 +1579,7 @@ class ComplexityRouter(CustomLogger):
         if escalated:
             signals = (*signals, "escalation")
         score_repr = f"{score:.3f}" if score is not None else "n/a"
+        served_tier = tier
         if self.config.adaptive:
             routed_model = self._soft_floor_pick(tier, user_message, request_kwargs)
             adaptive = self._ensure_adaptive_router()
@@ -1497,10 +1594,13 @@ class ComplexityRouter(CustomLogger):
                 f"signals={signals}, routed_model={routed_model}"
             )
         else:
-            routed_model = await self._pick_model_for_tier(tier, messages, resolved_messages, request_kwargs)
+            routed_model, served_tier = await self._pick_model_for_tier(
+                tier, messages, resolved_messages, request_kwargs
+            )
             verbose_router_logger.info(
-                f"ComplexityRouter: routing decision cause={outcome.cause}, tier={tier.value}, "
+                f"ComplexityRouter: routing decision cause={outcome.cause}, tier={served_tier.value}, "
                 f"score={score_repr}, signals={signals}, routed_model={routed_model}"
+                + (f", tier_fallback_from={tier.value}" if served_tier != tier else "")
             )
 
         classifier_model = (
@@ -1514,11 +1614,12 @@ class ComplexityRouter(CustomLogger):
             routing_decision=self._build_routing_decision(
                 routed_model=routed_model,
                 cause=outcome.cause,
-                tier=tier,
+                tier=served_tier,
                 score=score,
                 signals=signals,
                 escalation_keyword=escalation_keyword,
                 escalated=escalated,
                 classifier_model=classifier_model,
+                tier_fallback_from=tier if served_tier != tier else None,
             ),
         )

litellm/types/utils.py

@@ -2733,6 +2733,7 @@ class StandardLoggingRoutingDecision(TypedDict, total=False):
     escalation_keyword: str
     classifier_model: str
     escalated: bool
+    tier_fallback_from: str
     tier_boundaries: StandardLoggingRoutingDecisionTierBoundaries
 
 
@@ -2752,6 +2753,7 @@ DERIVED_ROUTING_DECISION_FIELDS: FrozenSet[str] = frozenset(
         "score",
         "classifier_model",
         "escalated",
+        "tier_fallback_from",
         "tier_boundaries",
     }
 )

tests/test_litellm/router_strategy/test_complexity_router.py

@@ -4738,3 +4738,175 @@ class TestClassifierTrustBoundary:
         assert system_message["content"] == _CLASSIFICATION_SYSTEM_RUBRIC
         assert hostile not in system_message["content"]
         assert hostile in user_message["content"]
+
+
+class StubRouter:
+    """Router stand-in that reports which model groups are servable.
+
+    `live` is the set of model groups with a healthy deployment; anything else is either
+    unknown to the router or entirely in cooldown, which the complexity router treats the
+    same way: it cannot serve this request now.
+    """
+
+    def __init__(self, live: set[str]):
+        self.live = live
+        self.cache = DualCache()
+        self.probed: list[str] = []
+
+    async def _async_get_healthy_deployments(self, model: str, parent_otel_span=None):
+        self.probed.append(model)
+        deployments = [{"model_info": {"id": f"{model}-0"}}] if model in self.live else []
+        return deployments, deployments
+
+
+LADDER_TIERS = {
+    "SIMPLE": ["simple-a", "simple-b"],
+    "MEDIUM": "medium-model",
+    "COMPLEX": "complex-model",
+    "REASONING": "reasoning-model",
+}
+
+
+def _ladder_router(live: set[str], **overrides):
+    return ComplexityRouter(
+        model_name="test-ladder-router",
+        litellm_router_instance=StubRouter(live),
+        complexity_router_config={"tiers": dict(LADDER_TIERS), "session_affinity": False, **overrides},
+    )
+
+
+async def _route(router, content: str):
+    return await router.async_pre_routing_hook(
+        model="test-model",
+        request_kwargs={},
+        messages=[{"role": "user", "content": content}],
+    )
+
+
+class TestTierFallbackLadder:
+    """A classified tier that cannot serve falls to its own peers first, then upward. It
+    never falls to a cheaper tier, because that would answer a hard request with the model
+    the classifier already rejected."""
+
+    @pytest.mark.asyncio
+    async def test_dead_pool_member_falls_to_a_peer_in_the_same_tier(self):
+        router = _ladder_router(live={"simple-b", "medium-model"})
+        for _ in range(8):
+            response = await _route(router, "What is 2+2?")
+            assert response.model == "simple-b"
+            assert response.routing_decision["tier"] == "SIMPLE"
+            assert "tier_fallback_from" not in response.routing_decision
+
+    @pytest.mark.asyncio
+    async def test_whole_tier_dead_bumps_up_one_tier_and_records_it(self):
+        router = _ladder_router(live={"medium-model", "complex-model"})
+        response = await _route(router, "What is 2+2?")
+        assert response.model == "medium-model"
+        decision = response.routing_decision
+        assert decision["tier"] == "MEDIUM"
+        assert decision["tier_fallback_from"] == "SIMPLE"
+
+    @pytest.mark.asyncio
+    async def test_climbs_past_several_dead_tiers(self):
+        router = _ladder_router(live={"reasoning-model"})
+        response = await _route(router, "What is 2+2?")
+        assert response.model == "reasoning-model"
+        assert response.routing_decision["tier"] == "REASONING"
+        assert response.routing_decision["tier_fallback_from"] == "SIMPLE"
+
+    @pytest.mark.asyncio
+    async def test_never_falls_to_a_cheaper_tier(self):
+        """REASONING is dead with nothing above it, and the cheaper tiers are alive. They
+        are still not used: they are what the classifier already ruled out. With no
+        default_model to fall to, the request stays on its own tier and lets the router
+        report the real failure, since a cooldown may have expired since the health read."""
+        router = _ladder_router(live={"simple-a", "medium-model", "complex-model"})
+        response = await _route(router, "Think step by step and reason through this carefully")
+        assert response.model == "reasoning-model"
+        assert response.routing_decision["tier"] == "REASONING"
+        assert "tier_fallback_from" not in response.routing_decision
+
+    @pytest.mark.asyncio
+    async def test_default_model_is_the_last_resort_not_the_first(self):
+        router = _ladder_router(live={"complex-model"}, default_model="fallback-model")
+        response = await _route(router, "What is 2+2?")
+        assert response.model == "complex-model"
+        assert response.routing_decision["tier_fallback_from"] == "SIMPLE"
+
+    @pytest.mark.asyncio
+    async def test_default_model_serves_when_the_ladder_is_exhausted(self):
+        router = _ladder_router(live=set(), default_model="fallback-model")
+        response = await _route(router, "What is 2+2?")
+        assert response.model == "fallback-model"
+
+    @pytest.mark.asyncio
+    async def test_a_healthy_tier_is_probed_and_no_higher_tier_is(self):
+        """The ladder stops at the first servable tier instead of pricing out every tier on
+        every request."""
+        router = _ladder_router(live={"simple-a", "simple-b"})
+        stub = router.litellm_router_instance
+        await _route(router, "What is 2+2?")
+        assert set(stub.probed) <= {"simple-a", "simple-b"}
+
+    @pytest.mark.asyncio
+    async def test_health_lookup_failure_keeps_the_classified_tier(self):
+        """A health view that cannot be read must not empty every tier and push traffic up
+        the ladder; it degrades to serving the configured pool."""
+
+        class BrokenRouter(StubRouter):
+            async def _async_get_healthy_deployments(self, model: str, parent_otel_span=None):
+                raise RuntimeError("cooldown cache unavailable")
+
+        router = ComplexityRouter(
+            model_name="test-ladder-router",
+            litellm_router_instance=BrokenRouter(live=set()),
+            complexity_router_config={"tiers": dict(LADDER_TIERS), "session_affinity": False},
+        )
+        response = await _route(router, "What is 2+2?")
+        assert response.model in {"simple-a", "simple-b"}
+        assert "tier_fallback_from" not in response.routing_decision
+
+    @pytest.mark.asyncio
+    async def test_an_explicitly_empty_pool_still_raises(self):
+        """Distinct from an unservable tier: an empty pool is a config error, and climbing
+        past it would hide the misconfiguration behind a more expensive model."""
+        router = ComplexityRouter(
+            model_name="test-ladder-router",
+            litellm_router_instance=StubRouter(live={"medium-model"}),
+            complexity_router_config={
+                "tiers": {"SIMPLE": [], "MEDIUM": "medium-model"},
+                "session_affinity": False,
+            },
+        )
+        with pytest.raises(ValueError, match="Empty model pool for tier SIMPLE"):
+            await _route(router, "What is 2+2?")
+
+    @pytest.mark.asyncio
+    async def test_plugins_run_against_the_tier_that_actually_serves(self):
+        """The ladder climbs before plugins run, so a plugin still vets whatever is served
+        rather than being handed a tier that cannot answer."""
+        seen: list[list[str]] = []
+
+        class RecordingPlugin:
+            async def run(self, context):
+                seen.append(list(context.candidate_models))
+                return context
+
+        router = _ladder_router(live={"complex-model"}, plugins=[RecordingPlugin()])
+        response = await _route(router, "What is 2+2?")
+        assert response.model == "complex-model"
+        assert seen == [["complex-model"]]
+
+    @pytest.mark.asyncio
+    async def test_a_plugin_narrowing_to_zero_still_fails_closed(self):
+        """Unchanged policy: a plugin refusing every candidate is a decision, so the ladder
+        must not climb past it and serve what the plugin just refused."""
+
+        class DenyAllPlugin:
+            async def run(self, context):
+                context.candidate_models = []
+                return context
+
+        router = _ladder_router(live={"simple-a", "medium-model"}, plugins=[DenyAllPlugin()])
+        with pytest.raises(ValueError, match="No candidate models left for tier SIMPLE"):
+            await _route(router, "What is 2+2?")