feat(complexity_router): opt-in modality-based capability routing for image requests (#39032)

litellm/litellm_core_utils/prompt_templates/common_utils.py

@@ -205,6 +205,41 @@ def is_non_content_values_set(message: AllMessageValues) -> bool:
     return any(message.get(key, None) is not None for key in message if key not in ignore_keys)
 
 
+_IMAGE_CONTENT_PART_TYPES: Final = frozenset({"image_url", "input_image", "image"})
+_IMAGE_SCAN_MAX_DEPTH: Final = 4
+
+
+def _content_parts_contain_image(parts: Sequence[object]) -> bool:
+    """Depth-bounded frontier walk over nested content lists, iterative because the repo bans
+    recursion; an Anthropic tool_result nests its image parts exactly one level down."""
+    frontier = parts  # rebind-ok: depth-bounded frontier walk
+    for _ in range(_IMAGE_SCAN_MAX_DEPTH):
+        if any(isinstance(part, Mapping) and part.get("type") in _IMAGE_CONTENT_PART_TYPES for part in frontier):
+            return True
+        frontier = tuple(  # rebind-ok: depth-bounded frontier walk
+            nested
+            for part in frontier
+            if isinstance(part, Mapping)
+            for content in (part.get("content"),)
+            if isinstance(content, list)
+            for nested in content
+        )
+        if not frontier:
+            return False
+    return False
+
+
+def request_contains_image_content(messages: Sequence[Mapping[str, object]]) -> bool:
+    """Whether any message carries an image content part, across the dialects that reach
+    pre-routing hooks untranslated: chat-completions ``image_url``, Responses ``input_image``,
+    and Anthropic Messages ``image``, including images nested inside ``tool_result`` blocks."""
+    return any(
+        isinstance(content, list) and _content_parts_contain_image(content)
+        for message in messages
+        for content in (message.get("content"),)
+    )
+
+
 def _audio_or_image_in_message_content(message: AllMessageValues) -> bool:
     """
     Checks if message content contains an image or audio

litellm/router_strategy/complexity_router/README.md

@@ -154,6 +154,9 @@ model_list:
         
         # Fallback model if tier cannot be determined
         default_model: gpt-4o
+
+        # Replace a routed model that cannot take image input (default: false)
+        modality_routing: true
 ```
 
 ## Usage
@@ -178,6 +181,25 @@ response = litellm.completion(
 
 ## Special Behaviors
 
+### Modality-based capability routing
+
+The classifier reads text alone, so a request carrying an image can classify cheap and land on a
+text-only model, which rejects it with a provider 400 no fallback catches. With
+`modality_routing: true`, one gate inspects every decided placement: when the routed model is
+explicitly declared `supports_vision: false` (deployment `model_info` first, the model cost map
+otherwise; unmapped names stay routable, and a multi-deployment group must accept on every
+deployment), the request is re-placed on the nearest HIGHER tier holding a capable model, with
+routing plugins still applied to the re-pick, then on `default_model` (never on plugin routers
+and never for a plan-floored decision), and otherwise rejected with a clear 400 naming the
+router. The walk only ever goes up, so a plan-mode floor cannot be undercut; a router whose only
+vision model sits below the decided tier gets the 400 and an actionable message instead.
+
+A same-tier re-pick keeps the decision's cause and adds `modality:image` to `signals`; a tier
+change or default takeover records `cause: modality_escalation` with the displaced placement
+(`modality_escalated_from:<TIER>` or `modality_displaced_default_model`). Escalations are never
+pinned by session affinity, and a KEPT session pin bypasses the gate entirely: a session pinned
+to a text-only model keeps it even when an image arrives.
+
 ### Heuristic-first chaining
 
 `classifier_type: heuristic_first` runs the local scorer on every request and only calls the LLM

litellm/router_strategy/complexity_router/complexity_router.py

@@ -30,6 +30,7 @@ from litellm.constants import EMPTY_MAPPING, RETURN_RAW_MODEL_NAME_METADATA_KEY
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.core_helpers import get_metadata_variable_name_from_kwargs
 from litellm.litellm_core_utils.internal_call_metadata import forwarded_internal_call_metadata
+from litellm.litellm_core_utils.prompt_templates.common_utils import request_contains_image_content
 from litellm.litellm_core_utils.sensitive_data_masker import mask_credentials_in_payload
 from litellm.llms.base_llm.base_utils import type_to_response_format_param
 from litellm.types.utils import (
@@ -738,6 +739,10 @@ def _decision_is_pinnable(decision: StandardLoggingRoutingDecision | None) -> bo
     size shrinks again the moment the client compacts: pinning the escalated tier would hold the
     session on the big-window model long after the oversized context that forced it is gone. The
     gate re-fires per request, so leaving these unpinned costs nothing but the classifier call.
+
+    A modality escalation is transient the same way: it describes what this one call carries (an
+    image), not what the session's traffic looks like, and pinning it would hold every following
+    text turn on the vision-capable model the image forced.
     """
     return decision is None or (
         decision.get("cause")
@@ -745,6 +750,7 @@ def _decision_is_pinnable(decision: StandardLoggingRoutingDecision | None) -> bo
             "default_model_fallback",
             "plan_mode",
             "housekeeping",
+            "modality_escalation",
         )
         and not decision.get("context_escalated")
     )
@@ -2274,6 +2280,175 @@ class ComplexityRouter(CustomLogger):
             return pinned_model
         return self.get_model_for_tier(escalated_tier)
 
+    def _model_accepts_image_input(self, model_name: str) -> bool:
+        """Whether a routed model or pool entry can serve an image request.
+
+        Resolved through the deployments that would actually serve the name; a name with no
+        deployment on the router is served by the SDK directly and is checked against the model
+        cost map itself. Only an explicit supports_vision false excludes, a deployment-level
+        model_info override first and the map otherwise, so unmapped custom names stay routable.
+
+        A multi-deployment group must accept on EVERY deployment: the router picks a deployment
+        inside the group after this gate runs, so a mixed group marked eligible could still hand
+        the image to its text-only member and fail with the exact 400 the gate exists to prevent.
+        """
+        from litellm.utils import is_vision_explicitly_disabled
+
+        def deployment_accepts(deployment: Mapping[str, Any]) -> bool:
+            declared: Final = (deployment.get("model_info") or EMPTY_MAPPING).get("supports_vision")
+            if declared is not None:
+                return declared is True
+            litellm_model: Final = (deployment.get("litellm_params") or EMPTY_MAPPING).get("model") or model_name
+            return not is_vision_explicitly_disabled(litellm_model)
+
+        deployments: Final = self.litellm_router_instance.get_model_list(model_name=model_name)
+        if not deployments:
+            return not is_vision_explicitly_disabled(model_name)
+        return all(deployment_accepts(deployment) for deployment in deployments)
+
+    def _modality_eligible_models(self) -> frozenset[str]:
+        """Every configured pool entry, plus default_model, that can serve an image request."""
+        names: Final = frozenset(entry for pool in self._tier_pools().values() for entry in pool) | frozenset(
+            name for name in (self.config.default_model,) if name
+        )
+        return frozenset(name for name in names if self._model_accepts_image_input(name))
+
+    async def _gate_response_modality(
+        self,
+        response: PreRoutingHookResponse,
+        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
+    ) -> PreRoutingHookResponse:
+        """Replace a routed model that cannot accept this request's image input.
+
+        The single modality owner, applied to the decided response at the hook's exits so every
+        routing path is covered uniformly. A KEPT session pin is exempt by design (its cause);
+        replacement picks and every other path are just responses. The re-placement walks
+        UPWARD-ONLY from the decision's tier (so a plan-mode floor can never be undercut), picks
+        through `_pick_model_for_tier` so routing plugins still apply, then falls to
+        default_model (never on plugin routers, and never on a plan-floored decision, since
+        default_model carries no tier guarantee), else raises the clear 400. The rewritten
+        decision keeps its cause on a same-tier repick and becomes modality_escalation when the
+        tier moved or default_model took over, with the displaced placement in signals.
+        """
+        decision: Final = response.routing_decision
+        if (
+            not self.config.modality_routing
+            or not resolved_messages
+            or response.model is None
+            or (decision is not None and decision.get("cause") == "session_affinity_pin")
+            or not request_contains_image_content(resolved_messages)
+            or self._model_accepts_image_input(response.model)
+        ):
+            return response
+        eligible: Final = self._modality_eligible_models()
+        names: Final = self.config.tier_names()
+        pools: Final = self._tier_pools()
+        decided: Final = decision.get("tier") if decision is not None else None
+        start: Final = names.index(decided) if isinstance(decided, str) and decided in names else 0
+        capable: Final = next(
+            (name for name in names[start:] if any(entry in eligible for entry in pools.get(name, ()))), None
+        )
+        if capable is not None:
+            new_tier: ComplexityTier | str | None = capable if self.config.has_custom_tiers else ComplexityTier(capable)
+            repick_messages: Final = list(resolved_messages)  # mutable-ok: the pick's param is list-typed
+            new_model = await self._pick_model_for_tier(
+                new_tier,
+                messages,
+                repick_messages,  # pyright: ignore[reportArgumentType]  # hook-resolved message dicts; the pick only reads them
+                request_kwargs,
+                allowed_models=tuple(entry for entry in pools.get(capable, ()) if entry in eligible),
+            )
+        elif self._modality_default_model_usable(request_kwargs, resolved_messages, eligible):
+            new_tier = None
+            new_model = self._placed_default_model()
+        else:
+            import litellm
+
+            raise litellm.BadRequestError(
+                message=(
+                    f"Auto-router {self.model_name} received a request with image input, but no model "
+                    f"at or above the decided tier accepts images and modality_routing is enabled. "
+                    f"Tiers checked: {', '.join(names[start:])}. Add a vision-capable model to a tier, "
+                    f"or set a vision-capable default_model, or remove the image content."
+                ),
+                model=self.model_name,
+                llm_provider="",
+            )
+        self._restamp_adaptive_choice(request_kwargs, response.model, new_model)
+        same_tier: Final = capable is not None and decided == capable
+        base_cause: Final = (decision.get("cause") if decision is not None else None) or "default_fallback"
+        displaced_default: Final = decided is None and response.model == self.config.default_model
+        markers: Final = (
+            "modality:image",
+            *((f"modality_escalated_from:{decided}",) if not same_tier and isinstance(decided, str) else ()),
+            *(("modality_displaced_default_model",) if not same_tier and displaced_default else ()),
+        )
+        old_signals: Final = tuple(decision.get("signals") or ()) if decision is not None else ()
+        new_decision: Final = self._build_routing_decision(
+            routed_model=new_model,
+            cause=base_cause if same_tier else "modality_escalation",
+            tier=new_tier,
+            score=decision.get("score") if decision is not None else None,
+            signals=(*old_signals, *markers),
+            matched_keyword=decision.get("matched_keyword") if decision is not None else None,
+            escalation_keyword=decision.get("escalation_keyword") if decision is not None else None,
+            escalated=bool(decision.get("escalated", False)) if decision is not None else False,
+            classifier_model=decision.get("classifier_model") if decision is not None else None,
+            classifier_cost=decision.get("classifier_cost") if decision is not None else None,
+            conversation_continuing=bool(decision.get("conversation_continuing", True))
+            if decision is not None
+            else True,
+            tier_litellm_params=self._litellm_params_for_model(new_tier, new_model),
+            context_escalation_original_tier=(
+                decision.get("context_escalation_original_tier") if decision is not None else None
+            ),
+        )
+        from litellm.types.router import PreRoutingHookResponse as HookResponse
+
+        return HookResponse(
+            model=new_model,
+            messages=response.messages,
+            litellm_params=self._litellm_params_for_model(new_tier, new_model),
+            routing_decision=new_decision,
+        )
+
+    def _modality_default_model_usable(
+        self,
+        request_kwargs: Mapping[str, object],
+        resolved_messages: Sequence[Mapping[str, object]] | None,
+        eligible: frozenset[str],
+    ) -> bool:
+        """default_model may serve a gated request only when it is configured, plugin-free
+        (it is never checked against the plugin pipeline), capability-eligible, and the turn
+        carries no plan-mode sentinel. The sentinel is re-detected here rather than read off
+        the decision record, because the record only marks turns the floor RAISED; a sentinel
+        turn already at or above the floor keeps its ordinary cause, and default_model carries
+        no tier the floor could vouch for on any sentinel turn."""
+        return (
+            bool(self.config.default_model)
+            and not self.config.plugins
+            and self.config.default_model in eligible
+            and self._matched_plan_mode_signal(request_kwargs, resolved_messages) is None
+        )
+
+    def _placed_default_model(self) -> str:
+        """The default_model behind a usable-default verdict; the raise is the type-level
+        proof, not a reachable path."""
+        model: Final = self.config.default_model
+        if model is None:
+            raise ValueError(f"Auto-router {self.model_name}: modality gate routed to an unset default_model")
+        return model
+
+    @staticmethod
+    def _restamp_adaptive_choice(request_kwargs: Mapping[str, object], old_model: str, new_model: str) -> None:
+        """The adaptive feedback loop reads its chosen-model marker from request metadata; a
+        gate rewrite must move the marker with the model or rewards land on the displaced one."""
+        metadata: Final = request_kwargs.get("metadata")
+        if isinstance(metadata, dict) and metadata.get("adaptive_router_chosen_model") == old_model:
+            metadata["adaptive_router_chosen_model"] = new_model
+
     def _lexical_tier_override(self, user_message: str) -> KeywordOverride | None:
         """When keyword_tier_rules match literally, the most-severe matched tier wins.
 
@@ -2655,25 +2830,30 @@ class ComplexityRouter(CustomLogger):
                     session_tier_litellm_params: Final = self._litellm_params_for_model(routed_pin_tier, routed_model)
                     has_original_messages: Final = messages is not None and len(messages) > 0
                     return self._with_session_deployment_affinity(
-                        PreRoutingHookResponse(
-                            model=routed_model,
-                            messages=messages if has_original_messages else None,
-                            litellm_params=session_tier_litellm_params,
-                            routing_decision=self._build_routing_decision(
-                                routed_model=routed_model,
-                                cause=cause,
-                                tier=routed_pin_tier,
-                                matched_keyword=pin_plan_sentinel if plan_floored else None,
-                                escalation_keyword=pin_escalation_keyword,
-                                escalated=escalated,
-                                conversation_continuing=conversation_continuing,
-                                tier_litellm_params=session_tier_litellm_params,
-                                context_escalation_original_tier=pin_context_original_tier,
+                        await self._gate_response_modality(
+                            PreRoutingHookResponse(
+                                model=routed_model,
+                                messages=messages if has_original_messages else None,
+                                litellm_params=session_tier_litellm_params,
+                                routing_decision=self._build_routing_decision(
+                                    routed_model=routed_model,
+                                    cause=cause,
+                                    tier=routed_pin_tier,
+                                    matched_keyword=pin_plan_sentinel if plan_floored else None,
+                                    escalation_keyword=pin_escalation_keyword,
+                                    escalated=escalated,
+                                    conversation_continuing=conversation_continuing,
+                                    tier_litellm_params=session_tier_litellm_params,
+                                    context_escalation_original_tier=pin_context_original_tier,
+                                ),
                             ),
+                            messages,
+                            resolved_messages,
+                            request_kwargs,
                         )
                     )
 
-        response: Final = await self._classify_and_route(
+        routed_response: Final = await self._classify_and_route(
             model=model,
             request_kwargs=request_kwargs,
             messages=messages,
@@ -2682,6 +2862,11 @@ class ComplexityRouter(CustomLogger):
             conversation_continuing=conversation_continuing,
             resolved_messages=resolved_messages,
         )
+        response: Final = (
+            await self._gate_response_modality(routed_response, messages, resolved_messages, request_kwargs)
+            if routed_response is not None
+            else None
+        )
         # Sentinel presence, not the plan_mode cause, gates the pin write: a plan-mode turn
         # classified at or above the floor keeps its ordinary cause, yet on an adaptive router
         # the hard floor constrained its pick, so pinning it would carry a plan-mode-shaped

litellm/router_strategy/complexity_router/config.py

@@ -848,6 +848,18 @@ class ComplexityRouterConfig(BaseModel):
             "drift plus the response tokens."
         ),
     )
+    modality_routing: bool = Field(
+        default=False,
+        description=(
+            "Route image-bearing requests only to models that can accept image input. The "
+            "classifier reads text alone, so an image request whose text classifies cheap "
+            "otherwise lands on a text-only model and fails with a provider 400. When enabled, "
+            "a routed model explicitly declared supports_vision false (deployment model_info "
+            "or the model cost map; unmapped names stay routable) is replaced by the nearest "
+            "HIGHER tier holding a capable model, then default_model, else a clear 400. A kept "
+            "session-affinity pin still wins even when an image arrives."
+        ),
+    )
 
     # Semantic (embedding) matching for keyword_tier_rules instead of literal text matching
     semantic_keyword_matching: bool = Field(

litellm/types/utils.py

@@ -2840,6 +2840,10 @@ RoutingDecisionCause = Literal[
     # never called. The matched sentinel rides in matched_keyword. Distinct from the keyword causes,
     # which are operator-authored rules; these sentinels ship with the router.
     "housekeeping",
+    # modality_routing replaced the decided placement: the request carries an image and the
+    # routed model does not accept image input, so the nearest higher capable tier or
+    # default_model served instead. The displaced placement rides in signals.
+    "modality_escalation",
     "session_affinity_pin",
     "session_affinity_escalation",
     # classification_mode 'user_turn': the request is an agent loop's continuation turn (no new

litellm/utils.py

@@ -2660,10 +2660,19 @@ def _is_explicitly_disabled_factory(model: str, custom_llm_provider: str | None,
     ``_supports_factory`` so caching, fallback, and normalisation improvements
     apply here automatically.
     """
+    from litellm.litellm_core_utils.get_llm_provider_logic import declared_authenticating_provider
+
     try:
-        model, custom_llm_provider, _, _ = litellm.get_llm_provider(
-            model=model, custom_llm_provider=custom_llm_provider
-        )
+        declared: Final = declared_authenticating_provider(model, custom_llm_provider)
+        if declared is not None:
+            model = model.removeprefix(
+                f"{declared}/"
+            )  # rebind-ok: mirrors get_llm_provider's split without its OAuth flow
+            custom_llm_provider = declared  # rebind-ok: same
+        else:
+            model, custom_llm_provider, _, _ = litellm.get_llm_provider(
+                model=model, custom_llm_provider=custom_llm_provider
+            )
         model_info: Final = _get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
         val: Final = model_info.get(key)
         if val is False:
@@ -2751,6 +2760,15 @@ def supports_computer_use(model: str, custom_llm_provider: str | None = None) ->
     )
 
 
+def is_vision_explicitly_disabled(model: str, custom_llm_provider: str | None = None) -> bool:
+    """True only when supports_vision is explicitly declared false for the model.
+
+    The opt-out mirror of :func:`supports_vision`: a missing declaration reads as not
+    disabled, so unknown or newly added models stay eligible for image routing.
+    """
+    return _is_explicitly_disabled_factory(model, custom_llm_provider, "supports_vision")
+
+
 def supports_vision(model: str, custom_llm_provider: str | None = None) -> bool:
     """
     Check if the given model supports vision and return a boolean value.

tests/test_litellm/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py

@@ -1433,3 +1433,50 @@ class TestFlattenTopLevelSchemaCombinators:
         flatten_top_level_schema_combinators(schema)
 
         assert schema == snapshot
+
+
+class TestRequestContainsImageContent:
+    """One detector for every dialect that reaches pre-routing hooks untranslated."""
+
+    @pytest.mark.parametrize(
+        "part",
+        [
+            {"type": "image_url", "image_url": {"url": "data:image/png;base64,aGk="}},
+            {"type": "input_image", "image_url": "data:image/png;base64,aGk="},
+            {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "aGk="}},
+            {
+                "type": "tool_result",
+                "tool_use_id": "tu_1",
+                "content": [{"type": "image", "source": {"type": "base64", "data": "aGk="}}],
+            },
+        ],
+    )
+    def test_detects_every_image_dialect_including_tool_results(self, part):
+        from litellm.litellm_core_utils.prompt_templates.common_utils import request_contains_image_content
+
+        messages = [{"role": "user", "content": [{"type": "text", "text": "hi"}, part]}]
+        assert request_contains_image_content(messages) is True
+
+    @pytest.mark.parametrize(
+        "messages",
+        [
+            [{"role": "user", "content": "plain string"}],
+            [{"role": "user", "content": [{"type": "text", "text": "hi"}]}],
+            [{"role": "user", "content": [{"type": "input_audio", "input_audio": {"data": "x"}}]}],
+            [{"role": "user", "content": [{"type": "tool_result", "content": [{"type": "text", "text": "ok"}]}]}],
+            [{"role": "user", "content": None}],
+            [],
+        ],
+    )
+    def test_ignores_text_audio_and_degenerate_shapes(self, messages):
+        from litellm.litellm_core_utils.prompt_templates.common_utils import request_contains_image_content
+
+        assert request_contains_image_content(messages) is False
+
+    def test_hostile_nesting_is_depth_bounded(self):
+        from litellm.litellm_core_utils.prompt_templates.common_utils import request_contains_image_content
+
+        nested: dict = {"type": "image", "source": {"type": "base64", "data": "aGk="}}
+        for _ in range(50):
+            nested = {"type": "tool_result", "content": [nested]}
+        assert request_contains_image_content([{"role": "user", "content": [nested]}]) is False

tests/test_litellm/router_strategy/test_complexity_router.py

@@ -10417,3 +10417,323 @@ class TestContextWindowEscalation:
 
         assert oversized["model_name"] == "big-model"
         assert small["model_name"] == "small-model"
+
+
+IMG_PART = {"type": "image_url", "image_url": {"url": "data:image/png;base64,aGk="}}
+PLAN_BODY = {
+    "messages": [{"role": "system", "content": [{"type": "text", "text": "Plan mode is active. Do not execute."}]}]
+}
+
+
+class TestModalityRouting:
+    """modality_routing: the response gate replaces a routed model that cannot take images."""
+
+    IMAGE_MESSAGE = [{"role": "user", "content": [{"type": "text", "text": "What color is this?"}, IMG_PART]}]
+    BASE_TIERS = {"SIMPLE": "text-cheap", "MEDIUM": "vision-mid", "COMPLEX": "vision-big"}
+    BASE_VISION = {"text-cheap": False, "vision-mid": True, "vision-big": True, "vision-default": True}
+
+    @staticmethod
+    def _router(mock_router_instance, config, vision_by_model):
+        """vision_by_model: model name -> True/False (deployment model_info) or None (undeclared)."""
+
+        def get_model_list(model_name=None):
+            if model_name not in vision_by_model:
+                return []
+            declared = vision_by_model[model_name]
+            return [
+                {
+                    "model_name": model_name,
+                    "litellm_params": {"model": f"openai/unmapped-{model_name}"},
+                    "model_info": {} if declared is None else {"supports_vision": declared},
+                }
+            ]
+
+        mock_router_instance.get_model_list = get_model_list
+        return ComplexityRouter(
+            model_name="modality-test-router",
+            litellm_router_instance=mock_router_instance,
+            complexity_router_config=config,
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "config_extra, vision, send_image, expected_model, expect_marker",
+        [
+            ({}, {"text-cheap": False}, True, "text-cheap", False),
+            ({"modality_routing": True}, {"text-cheap": False}, False, "text-cheap", False),
+            ({"modality_routing": True}, {"text-cheap": None}, True, "text-cheap", False),
+        ],
+        ids=["flag_off", "no_image", "undeclared_model_stays_routable"],
+    )
+    async def test_gate_leaves_ungated_requests_untouched(
+        self, mock_router_instance, config_extra, vision, send_image, expected_model, expect_marker
+    ):
+        router = self._router(mock_router_instance, {"tiers": dict(self.BASE_TIERS), **config_extra}, vision)
+        request = self.IMAGE_MESSAGE if send_image else [{"role": "user", "content": "What color is the sky?"}]
+        result = await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=request)
+        assert result.model == expected_model
+        assert result.routing_decision["cause"] == "heuristic_scorer"
+        assert ("modality:image" in (result.routing_decision.get("signals") or ())) is expect_marker
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "part",
+        [
+            IMG_PART,
+            {"type": "input_image", "image_url": "data:image/png;base64,aGk="},
+            {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "aGk="}},
+            {"type": "tool_result", "tool_use_id": "tu_1", "content": [dict(IMG_PART, type="image")]},
+        ],
+        ids=["image_url", "input_image", "anthropic_image", "tool_result_nested"],
+    )
+    async def test_every_image_dialect_escalates(self, mock_router_instance, part):
+        router = self._router(
+            mock_router_instance, {"tiers": dict(self.BASE_TIERS), "modality_routing": True}, dict(self.BASE_VISION)
+        )
+        message = [{"role": "user", "content": [{"type": "text", "text": "What color is this?"}, part]}]
+        result = await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=message)
+        assert result.model == "vision-mid"
+        assert result.routing_decision["cause"] == "modality_escalation"
+        assert "modality_escalated_from:SIMPLE" in result.routing_decision["signals"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "path, expected_model, expected_cause",
+        [
+            ("classifier_escalates", "vision-mid", "modality_escalation"),
+            ("same_tier_repick_keeps_cause", "vision-cheap", "heuristic_scorer"),
+            ("keyword_tier_escalates", "vision-mid", "modality_escalation"),
+            ("no_ask_capable_default_kept", "vision-default", "default_fallback"),
+            ("no_ask_text_default_displaced", "vision-mid", "modality_escalation"),
+            ("custom_tiers_walk", "premium-model", "modality_escalation"),
+            ("pin_kept_bypasses", "text-cheap", "session_affinity_pin"),
+            ("pin_replacement_gated", "vision-big", "modality_escalation"),
+            ("adaptive_pick_rewritten", "vision-mid", "modality_escalation"),
+        ],
+    )
+    async def test_placements_across_decision_paths(self, mock_router_instance, path, expected_model, expected_cause):
+        config = {"tiers": dict(self.BASE_TIERS), "modality_routing": True}
+        vision = dict(self.BASE_VISION)
+        request_kwargs = {}
+        messages = self.IMAGE_MESSAGE
+        if path == "same_tier_repick_keeps_cause":
+            config["tiers"]["SIMPLE"] = ["text-cheap", "vision-cheap"]
+            vision["vision-cheap"] = True
+            with patch(  # test-quality-ok: the mixed-pool repick is unreachable deterministically without pinning the first random pick
+                "litellm.router_strategy.complexity_router.complexity_router.random.choice",
+                side_effect=lambda pool: sorted(pool)[0],
+            ):
+                router = self._router(mock_router_instance, config, vision)
+                result = await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=messages)
+            assert result.model == expected_model
+            assert result.routing_decision["cause"] == expected_cause
+            assert result.routing_decision["signals"][-1] == "modality:image"
+            return
+        if path == "keyword_tier_escalates":
+            config["keyword_tier_rules"] = [{"keywords": ["quick lookup"], "tier": "SIMPLE"}]
+            messages = [
+                {"role": "user", "content": [{"type": "text", "text": "quick lookup: what is this?"}, IMG_PART]}
+            ]
+        elif path == "no_ask_capable_default_kept":
+            config["default_model"] = "vision-default"
+            messages = [{"role": "user", "content": [IMG_PART]}]
+        elif path == "no_ask_text_default_displaced":
+            config["default_model"] = "text-default"
+            vision["text-default"] = False
+            messages = [{"role": "user", "content": [IMG_PART]}]
+        elif path == "custom_tiers_walk":
+            config = {
+                "classifier_type": "llm",
+                "classifier_llm_config": {"model": "gpt-4o-mini"},
+                "fallback_tier": "cheap",
+                "tier_definitions": [
+                    {"name": "cheap", "description": "trivial asks"},
+                    {"name": "premium", "description": "hard asks"},
+                ],
+                "tiers": {"cheap": "cheap-model", "premium": "premium-model"},
+                "keyword_tier_rules": [{"keywords": ["quick lookup"], "tier": "cheap"}],
+                "modality_routing": True,
+            }
+            vision = {"cheap-model": False, "premium-model": True}
+            messages = [
+                {"role": "user", "content": [{"type": "text", "text": "quick lookup: what is this?"}, IMG_PART]}
+            ]
+        elif path in ("pin_kept_bypasses", "pin_replacement_gated"):
+            cache = AsyncMock()
+            cache.async_get_cache = AsyncMock(return_value={"model": "text-cheap", "tier": "SIMPLE"})
+            mock_router_instance.cache = cache
+            config["session_affinity"] = True
+            request_kwargs = {"metadata": {"session_id": "s1"}}
+            if path == "pin_replacement_gated":
+                config["tiers"]["MEDIUM"] = "text-mid"
+                vision["text-mid"] = False
+                messages = [
+                    {"role": "user", "content": [{"type": "text", "text": "LITELLM ESCALATE describe this"}, IMG_PART]}
+                ]
+        elif path == "adaptive_pick_rewritten":
+            config["adaptive"] = True
+            mock_router_instance.model_list = []
+            mock_router_instance.model_name_to_deployment_indices = {}
+        router = self._router(mock_router_instance, config, vision)
+        result = await router.async_pre_routing_hook(model="m", request_kwargs=request_kwargs, messages=messages)
+        assert result.model == expected_model
+        assert result.routing_decision["cause"] == expected_cause
+        if path == "adaptive_pick_rewritten":
+            assert request_kwargs["metadata"]["adaptive_router_chosen_model"] == expected_model
+
+    @pytest.mark.asyncio
+    async def test_plan_floored_decision_never_falls_to_default_model(self, mock_router_instance):
+        """An upward-only walk cannot undercut the floor; default_model must not either."""
+        config = {
+            "tiers": {"SIMPLE": "vision-cheap", "MEDIUM": "text-mid"},
+            "default_model": "vision-default",
+            "plan_mode_min_tier": "MEDIUM",
+            "modality_routing": True,
+        }
+        vision = {"vision-cheap": True, "text-mid": False, "vision-default": True}
+        router = self._router(mock_router_instance, config, vision)
+        with pytest.raises(litellm.BadRequestError, match="no model"):
+            await router.async_pre_routing_hook(
+                model="m",
+                request_kwargs={"proxy_server_request": {"body": PLAN_BODY}},
+                messages=[{"role": "user", "content": [{"type": "text", "text": "plan this"}, IMG_PART]}],
+            )
+
+    @pytest.mark.asyncio
+    async def test_at_floor_plan_turn_never_falls_to_default_model(self, mock_router_instance):
+        """A sentinel turn whose classified tier already satisfies the floor keeps its ordinary
+        cause, so the record carries no floor marker; the default arm must still refuse it."""
+        config = {
+            "tiers": {"SIMPLE": "text-a", "MEDIUM": "text-b"},
+            "default_model": "vision-default",
+            "plan_mode_min_tier": "SIMPLE",
+            "modality_routing": True,
+        }
+        vision = {"text-a": False, "text-b": False, "vision-default": True}
+        router = self._router(mock_router_instance, config, vision)
+        with pytest.raises(litellm.BadRequestError, match="no model"):
+            await router.async_pre_routing_hook(
+                model="m",
+                request_kwargs={"proxy_server_request": {"body": PLAN_BODY}},
+                messages=[{"role": "user", "content": [{"type": "text", "text": "plan this"}, IMG_PART]}],
+            )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "default_model, default_vision, expect_error",
+        [(None, None, True), ("text-default", False, True), ("vision-default", True, False)],
+        ids=["no_default", "text_only_default", "vision_default_serves"],
+    )
+    async def test_no_capable_tier_above_uses_default_or_rejects(
+        self, mock_router_instance, default_model, default_vision, expect_error
+    ):
+        config = {"tiers": {"SIMPLE": "text-cheap", "COMPLEX": "text-big"}, "modality_routing": True}
+        vision = {"text-cheap": False, "text-big": False}
+        if default_model is not None:
+            config["default_model"] = default_model
+            vision[default_model] = default_vision
+        router = self._router(mock_router_instance, config, vision)
+        if expect_error:
+            with pytest.raises(litellm.BadRequestError, match="no model"):
+                await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=self.IMAGE_MESSAGE)
+            return
+        result = await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=self.IMAGE_MESSAGE)
+        assert result.model == "vision-default"
+        assert result.routing_decision["cause"] == "modality_escalation"
+        assert "modality_escalated_from:SIMPLE" in result.routing_decision["signals"]
+
+    @pytest.mark.asyncio
+    async def test_mixed_deployment_group_is_treated_text_only(self, mock_router_instance):
+        def get_model_list(model_name=None):
+            declared = {"mixed-group": [True, False], "vision-big": [True]}.get(model_name)
+            if declared is None:
+                return []
+            return [
+                {
+                    "model_name": model_name,
+                    "litellm_params": {"model": f"openai/unmapped-{model_name}-{i}"},
+                    "model_info": {"supports_vision": accepts},
+                }
+                for i, accepts in enumerate(declared)
+            ]
+
+        mock_router_instance.get_model_list = get_model_list
+        router = ComplexityRouter(
+            model_name="modality-test-router",
+            litellm_router_instance=mock_router_instance,
+            complexity_router_config={
+                "tiers": {"SIMPLE": "mixed-group", "COMPLEX": "vision-big"},
+                "modality_routing": True,
+            },
+        )
+        result = await router.async_pre_routing_hook(model="m", request_kwargs={}, messages=self.IMAGE_MESSAGE)
+        assert result.model == "vision-big"
+        assert result.routing_decision["cause"] == "modality_escalation"
+
+    @pytest.mark.asyncio
+    async def test_continuation_turn_screenshot_escalates_past_the_held_model(self, mock_router_instance):
+        """classification_mode user_turn replays the held model on continuation turns; a
+        continuation carrying a screenshot must still be re-placed when that model is text-only."""
+        mock_router_instance.cache = DualCache()
+        config = {
+            "tiers": dict(self.BASE_TIERS),
+            "classification_mode": "user_turn",
+            "modality_routing": True,
+        }
+        router = self._router(mock_router_instance, config, dict(self.BASE_VISION))
+        first = await router.async_pre_routing_hook(
+            model="m",
+            request_kwargs={"metadata": {"session_id": "cont-1"}},
+            messages=[{"role": "user", "content": "hi there"}],
+        )
+        assert first.model == "text-cheap"
+        continuation = [
+            {"role": "user", "content": "hi there"},
+            {"role": "assistant", "content": [{"type": "tool_use", "id": "tu_1", "name": "screenshot", "input": {}}]},
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": "tu_1",
+                        "content": [{"type": "image", "source": {"type": "base64", "data": "aGk="}}],
+                    }
+                ],
+            },
+        ]
+        second = await router.async_pre_routing_hook(
+            model="m", request_kwargs={"metadata": {"session_id": "cont-1"}}, messages=continuation
+        )
+        assert second.model == "vision-mid"
+        assert second.routing_decision["cause"] == "modality_escalation"
+        assert "modality_escalated_from:SIMPLE" in second.routing_decision["signals"]
+
+    @pytest.mark.asyncio
+    async def test_rewrite_carries_the_context_escalation_record(self, mock_router_instance):
+        """A context-window escalation and a modality re-place are separate facts on one
+        record; rewriting for the image must not drop the sibling gate's fields."""
+        from litellm.types.router import PreRoutingHookResponse
+
+        router = self._router(
+            mock_router_instance,
+            {"tiers": dict(self.BASE_TIERS), "modality_routing": True},
+            dict(self.BASE_VISION),
+        )
+        decision = router._build_routing_decision(
+            routed_model="text-cheap",
+            cause="heuristic_scorer",
+            tier=ComplexityTier.SIMPLE,
+            context_escalation_original_tier=ComplexityTier.SIMPLE,
+        )
+        response = PreRoutingHookResponse(model="text-cheap", messages=None, routing_decision=decision)
+        rewritten = await router._gate_response_modality(response, None, self.IMAGE_MESSAGE, {})
+        assert rewritten.model == "vision-mid"
+        assert rewritten.routing_decision["cause"] == "modality_escalation"
+        assert rewritten.routing_decision["context_escalated"] is True
+        assert rewritten.routing_decision["context_escalation_original_tier"] == "SIMPLE"
+
+    def test_modality_escalation_is_never_pinnable(self):
+        from litellm.router_strategy.complexity_router.complexity_router import _decision_is_pinnable
+
+        assert _decision_is_pinnable({"cause": "modality_escalation"}) is False
+        assert _decision_is_pinnable({"cause": "heuristic_scorer"}) is True

tests/test_litellm/test_utils.py

@@ -5765,3 +5765,33 @@ class TestHuggingFaceConfigFetch:
         assert _get_max_position_embeddings("some-org/some-model") == 512
         request_timeout = hf_config_route.calls.last.request.extensions["timeout"]
         assert request_timeout["read"] == HF_CONFIG_FETCH_TIMEOUT_SECONDS
+
+
+class TestIsVisionExplicitlyDisabled:
+    """github_copilot and chatgpt run an OAuth device flow inside get_llm_provider; the
+    explicit-disable lookup must adopt the declared prefix instead of resolving it, exactly
+    as _supports_factory does, or a capability check on a copilot deployment blocks routing
+    on a device-code prompt."""
+
+    @pytest.mark.parametrize("model", ["github_copilot/gpt-4o", "chatgpt/gpt-5"])
+    def test_never_resolves_an_authenticating_prefix(self, model, monkeypatch):
+        from litellm.utils import is_vision_explicitly_disabled
+
+        lookups: list = []
+
+        def _record(*args, **kwargs):
+            lookups.append((args, kwargs))
+            raise RuntimeError("provider resolution must not run for an authenticating provider")
+
+        monkeypatch.setattr(litellm, "get_llm_provider", _record)
+
+        assert is_vision_explicitly_disabled(model) is False
+        assert lookups == []
+
+    def test_explicit_false_detected_and_absent_reads_enabled(self):
+        from litellm.utils import is_vision_explicitly_disabled
+
+        assert (
+            is_vision_explicitly_disabled("fireworks_ai/accounts/fireworks/models/deepseek-v4-flash-0731") is True
+        )
+        assert is_vision_explicitly_disabled("anthropic/claude-sonnet-4-5") is False

ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/RoutingDecisionCard.test.tsx

@@ -186,6 +186,12 @@ describe("RoutingDecisionCard", () => {
     expect(screen.queryByText("housekeeping")).not.toBeInTheDocument();
   });
 
+  it("labels a modality escalation instead of showing the raw cause token", () => {
+    render(<RoutingDecisionCard decision={{ ...heuristic, cause: "modality_escalation" }} />);
+    expect(screen.getByText("Escalated for image input")).toBeInTheDocument();
+    expect(screen.queryByText("modality_escalation")).not.toBeInTheDocument();
+  });
+
   it("shows the escalation keyword", () => {
     render(
       <RoutingDecisionCard decision={{ ...heuristic, escalated: true, escalation_keyword: "LITELLM ESCALATE" }} />,

ui/litellm-dashboard/src/components/view_logs/LogDetailsDrawer/RoutingDecisionCard.tsx

@@ -90,6 +90,7 @@ const CONSTANT_CAUSE_LABELS: Record<string, string> = {
   session_affinity_pin: "Pinned to session",
   session_affinity_escalation: "Escalated from session pin",
   user_turn_continuation: "Continuation turn, classifier skipped",
+  modality_escalation: "Escalated for image input",
   quality_tier: "Quality tier mapping",
   bandit: "Adaptive bandit",
   default_fallback: "Default model, no route matched",

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

@@ -34463,6 +34463,12 @@ export interface components {
              * @default 0.5
              */
             match_threshold: number;
+            /**
+             * Modality Routing
+             * @description Route image-bearing requests only to models that can accept image input. The classifier reads text alone, so an image request whose text classifies cheap otherwise lands on a text-only model and fails with a provider 400. When enabled, a routed model explicitly declared supports_vision false (deployment model_info or the model cost map; unmapped names stay routable) is replaced by the nearest HIGHER tier holding a capable model, then default_model, else a clear 400. A kept session-affinity pin still wins even when an image arrives.
+             * @default false
+             */
+            modality_routing: boolean;
             /**
              * Plan Mode Min Tier
              * @description When set, requests carrying a coding-agent plan-mode sentinel (Claude Code plan mode, VS Code Copilot Plan mode, Copilot CLI's exit_plan_mode tool) are routed to at least this tier: the classified tier still wins when it is higher, and the floor also overrides a session-affinity pin to a lower tier for exactly the turns carrying the sentinel, without rewriting the pin -- the first turn after plan mode exits routes as if plan mode had never happened. Names a built-in tier, or with tier_definitions set, one of the defined tier names (list order is ascending severity, same as keyword_tier_rules). Unset disables detection entirely. The sentinels ride in client-injected prompt text, so a caller who pastes one can spend up to this tier's models -- never down, and never outside the configured pools.
@@ -35620,7 +35626,7 @@ export interface components {
              * Cause
              * @enum {string}
              */
-            cause?: "heuristic_scorer" | "reasoning_override" | "llm_classifier" | "heuristic_first_short_circuit" | "classifier_plugin" | "classifier_fallback" | "default_model_fallback" | "literal_keyword_match" | "semantic_keyword_match" | "plan_mode" | "housekeeping" | "session_affinity_pin" | "session_affinity_escalation" | "user_turn_continuation" | "default_fallback" | "keyword" | "quality_tier" | "bandit";
+            cause?: "heuristic_scorer" | "reasoning_override" | "llm_classifier" | "heuristic_first_short_circuit" | "classifier_plugin" | "classifier_fallback" | "default_model_fallback" | "literal_keyword_match" | "semantic_keyword_match" | "plan_mode" | "housekeeping" | "modality_escalation" | "session_affinity_pin" | "session_affinity_escalation" | "user_turn_continuation" | "default_fallback" | "keyword" | "quality_tier" | "bandit";
             /** Classifier Cost */
             classifier_cost?: number;
             /** Classifier Model */