feat(otel): trace auto-router configuration and classifier failures (#44926)

litellm/integrations/opentelemetry.py

@@ -32,6 +32,7 @@ from litellm.integrations.otel.model.db_endpoint import db_span_attributes
 from litellm.integrations.otel.model.metadata import flatten_metadata
 from litellm.integrations.otel.model.semconv import LiteLLM, Metric
 from litellm.integrations.otel.plumbing.otlp_tls import resolve_otlp_http_tls
+from litellm.integrations.otel.routing import routing_decision_attributes
 from litellm.litellm_core_utils.internal_call_metadata import is_unbilled_non_inference_call_from_params
 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
 from litellm.litellm_core_utils.secret_redaction import redact_string
@@ -2407,6 +2408,8 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
             metadata: Final = standard_logging_payload["metadata"]
             for key, value in metadata.items():
                 self.safe_set_attribute(span=span, key=f"metadata.{key}", value=value)
+            decision: Final = metadata.get("routing_decision")
+            span.set_attributes(routing_decision_attributes(decision))
 
             # get hidden params
             hidden_params: Final = getattr(standard_logging_payload, "hidden_params", None) or (

litellm/integrations/otel/logger.py

@@ -54,6 +54,7 @@ from litellm.integrations.otel.model.spans import SpanRole, span_role_for_servic
 from litellm.integrations.otel.model.utils import to_ns
 from litellm.integrations.otel.plumbing.context import (
     active_phase,
+    active_phase_span,
     is_recordable_span,
     mcp_message_transport_span,
     post_response_root,
@@ -778,6 +779,11 @@ class OpenTelemetryV2(CustomLogger):
         if is_recordable_span(span):
             span.add_event(name, attributes)
 
+    def set_phase_attributes(self, attributes: Mapping[str, str | int | float | bool]) -> None:
+        span: Final = active_phase_span()
+        if span is not None and is_recordable_span(span):
+            span.set_attributes(attributes)
+
     async def async_pre_call_hook(
         self,
         user_api_key_dict: "UserAPIKeyAuth",
@@ -1042,6 +1048,12 @@ def phase_event(name: str, attributes: Mapping[str, str | int] | None = None) ->
         logger.add_phase_event(name, attributes)
 
 
+def phase_attributes(attributes: Mapping[str, str | int | float | bool]) -> None:
+    logger: Final = _registered_v2_logger()
+    if logger is not None:
+        logger.set_phase_attributes(attributes)
+
+
 def build_otel_v2_logger(
     config: OpenTelemetryV2Config,
     callback_name: str | None = None,

litellm/integrations/otel/mappers/genai.py

@@ -174,7 +174,7 @@ class GenAIMapper:
                 return {}
 
     def _llm_call(self, data: LLMCallSpanData) -> AttributeMap:
-        attrs: Final = collect(self._LLM_CALL_ATTRS, data)
+        attrs: Final = {**collect(self._LLM_CALL_ATTRS, data), **data.routing_attributes}
         if data.tools:
             attrs[LiteLLM.TOOLS_DECLARED] = len(data.tools)
             attrs.update(

litellm/integrations/otel/model/payloads.py

@@ -30,6 +30,7 @@ from litellm.integrations.otel.model.utils import (
     as_str_mapping,
     as_str_tuple,
 )
+from litellm.integrations.otel.routing import RoutingAttributeValue, routing_decision_attributes
 
 # ``RequestIdentity`` and the request-metadata translation now live in
 # :mod:`metadata`; re-exported here so existing ``model.payloads`` imports keep
@@ -436,6 +437,7 @@ class LLMCallSpanData:
     trace: TraceControls = field(default_factory=TraceControls)
     session_id: str | None = None
     embedding_output: EmbeddingOutput | None = None
+    routing_attributes: Mapping[str, RoutingAttributeValue] = field(default_factory=lambda: MappingProxyType({}))
 
     @classmethod
     def from_standard_logging_payload(
@@ -453,6 +455,8 @@ class LLMCallSpanData:
         # model split, the response model, api base, and identity all come from
         # here rather than being re-derived from the raw payload dicts.
         context: Final = RequestContext.from_standard_logging_payload(payload)
+        metadata: Final = payload.get("metadata")
+        decision: Final = metadata["routing_decision"] if metadata and "routing_decision" in metadata else None
         # Normalize ``response`` to a dict once so the content/id reads below are a
         # plain ``.get`` — no repeated ``isinstance`` guards.
         raw_response: Final = payload.get("response")
@@ -496,6 +500,7 @@ class LLMCallSpanData:
             trace=trace or TraceControls(),
             session_id=session_id or None,
             embedding_output=embedding_output if capture_content else None,
+            routing_attributes=routing_decision_attributes(decision),
         )
 
 

litellm/integrations/otel/routing.py

@@ -0,0 +1,57 @@
+from collections.abc import Mapping
+from types import MappingProxyType
+from typing import Final, TypeAlias
+
+RoutingAttributeValue: TypeAlias = str | int | float | bool
+
+_ROUTING_SCALAR_FIELDS: Final = frozenset(
+    {
+        "router_model_name",
+        "router_type",
+        "router_config_id",
+        "router_config_updated_at",
+        "router_config_fingerprint",
+        "routed_model",
+        "cause",
+        "tier",
+        "tier_label",
+        "request_type",
+        "score",
+        "classifier_model",
+        "classifier_cost",
+        "classifier_failure_reason",
+        "classifier_error_type",
+        "classifier_confidence",
+        "classifier_primary_rule",
+        "classifier_capability_boundary",
+        "classifier_p_solve",
+        "classifier_calibrated_p_solve",
+        "classifier_calibration_version",
+        "classifier_efficient_p_solve",
+        "classifier_capable_p_solve",
+        "classifier_calibrated_efficient_p_solve",
+        "classifier_calibrated_capable_p_solve",
+        "classifier_max_quality_gap",
+        "classifier_prompt_version",
+        "classifier_threshold",
+        "escalated",
+        "context_escalated",
+        "context_escalation_original_tier",
+        "reasoning_override_min_score",
+        "conversation_continuing",
+        "savings_baseline_model",
+        "savings_baseline_deployment_id",
+    }
+)
+
+
+def routing_decision_attributes(decision: Mapping[str, object] | None) -> Mapping[str, RoutingAttributeValue]:
+    if not isinstance(decision, Mapping):
+        return MappingProxyType({})
+    return MappingProxyType(
+        {
+            f"litellm.routing.{key}": value
+            for key in _ROUTING_SCALAR_FIELDS
+            if isinstance(value := decision.get(key), (str, int, float, bool))
+        }
+    )

litellm/integrations/otel/runtime.py

@@ -7,7 +7,9 @@ V2 is not the active logger — so a call site can wrap a request phase or seed
 identity unconditionally.
 """
 
-from collections.abc import Callable, Iterator, Mapping
+from __future__ import annotations
+
+from collections.abc import Callable, Generator, Mapping
 from contextlib import AbstractContextManager, contextmanager
 from functools import cache
 from typing import TYPE_CHECKING, Final, TypeAlias
@@ -16,11 +18,18 @@ if TYPE_CHECKING:
     from opentelemetry.trace import Span
 
 PhaseEventAttributes: TypeAlias = Mapping[str, str | int]
+PhaseAttributes: TypeAlias = Mapping[str, str | int | float | bool]
 
 
 @cache
 def _otel_runtime() -> (
-    "tuple[Callable[[str], AbstractContextManager[Span | None]], Callable[..., None], Callable[[str, PhaseEventAttributes | None], None]] | None"
+    tuple[
+        Callable[[str], AbstractContextManager[Span | None]],
+        Callable[..., None],
+        Callable[[str, PhaseEventAttributes | None], None],
+        Callable[[PhaseAttributes], None],
+    ]
+    | None
 ):
     """Resolve the SDK-backed hooks once and cache the outcome, absence included.
 
@@ -32,11 +41,11 @@ def _otel_runtime() -> (
         from litellm.integrations.otel import logger
     except Exception:
         return None
-    return (logger.phase_span, logger.seed_request_identity, logger.phase_event)
+    return (logger.phase_span, logger.seed_request_identity, logger.phase_event, logger.phase_attributes)
 
 
 @contextmanager
-def phase_span(name: str) -> "Iterator[Span | None]":
+def phase_span(name: str) -> Generator[Span | None]:
     """Run a request phase inside a live active span so its DB/service calls nest.
 
     Yields ``None`` (a plain no-op) when the OTel SDK is unavailable or V2 is not
@@ -64,3 +73,10 @@ def seed_request_identity(user_api_key_dict: object, model: object = None) -> No
     if runtime is None:
         return
     runtime[1](user_api_key_dict, model=model)
+
+
+def phase_attributes(attributes: PhaseAttributes) -> None:
+    runtime: Final = _otel_runtime()
+    if runtime is None:
+        return
+    runtime[3](attributes)

litellm/router.py

@@ -73,7 +73,8 @@ from litellm.constants import (
 )
 from litellm.integrations.custom_guardrail import is_guardrail_intervention
 from litellm.integrations.custom_logger import CustomLogger
-from litellm.integrations.otel.runtime import phase_event, phase_span
+from litellm.integrations.otel.routing import routing_decision_attributes
+from litellm.integrations.otel.runtime import phase_attributes, phase_event, phase_span
 from litellm.litellm_core_utils.asyncify import run_async_function
 from litellm.litellm_core_utils.core_helpers import (
     _get_parent_otel_span_from_kwargs,
@@ -162,6 +163,7 @@ from litellm.router_utils.add_retry_fallback_headers import (
 )
 from litellm.router_utils.auto_router_model_naming import (
     AUTO_ROUTER_MODEL_PREFIX,
+    STRATEGY_ROUTER_PARAM_FIELDS,
     GatedAutoRouterCapability,
     capability_limit_violation,
     claimed_capabilities,
@@ -8724,6 +8726,19 @@ class Router:
         breadcrumbs: Final = (*kept_breadcrumbs, attempt_record)
         earlier: Final = request_metadata.get("request_retry_count")
         request_retry_count: Final = (earlier if type(earlier) is int and 0 <= earlier else 0) + 1
+        decision: Final = request_metadata.get("routing_decision")
+        router_name: Final = decision.get("router_model_name") if isinstance(decision, Mapping) else None
+        if isinstance(router_name, str):
+            phase_event(
+                "litellm.routing.retry",
+                {
+                    "litellm.routing.router_model_name": router_name,
+                    "litellm.retry.count": request_retry_count,
+                    "error.type": type(e).__name__,
+                    **({"litellm.deployment.model_group": model_group} if isinstance(model_group, str) else {}),
+                    **({"litellm.deployment.id": deployment_id} if isinstance(deployment_id, str) else {}),
+                },
+            )
         kwargs[_metadata_var]["previous_models"] = breadcrumbs  # rebind-ok: the logging object already holds this dict
         kwargs[_metadata_var]["request_retry_count"] = request_retry_count  # rebind-ok: same dict, read by the cap
         return kwargs
@@ -9572,9 +9587,29 @@ class Router:
             )
         registry[deployment.model_name] = [
             *registry.get(deployment.model_name, []),
-            TaggedPreRoutingStrategy(tags=tags, strategy=strategy),
+            TaggedPreRoutingStrategy(
+                tags=tags,
+                strategy=strategy,
+                definition_fingerprint=self._routing_definition_fingerprint(deployment),
+                deployment=deployment,
+            ),
         ]
 
+    @staticmethod
+    def _routing_definition_fingerprint(deployment: Deployment) -> str | None:
+        """Identify configured routing fields, excluding external files and live strategy state."""
+        try:
+            definition: Final = json.dumps(
+                deployment.litellm_params.model_dump(
+                    include=STRATEGY_ROUTER_PARAM_FIELDS | {"model", "tags"}, exclude_none=True, warnings=False
+                ),
+                sort_keys=True,
+                default=Router._json_default_stable_id,
+            )
+        except (TypeError, ValueError):
+            return None
+        return Router.generate_model_id(deployment.model_name, {"routing_definition": definition})
+
     @staticmethod
     def _unregister_pre_routing_strategy(
         registry: dict[str, list[TaggedPreRoutingStrategy[_PreRoutingStrategyT]]],
@@ -9653,7 +9688,12 @@ class Router:
                 if adaptive_router is not None:
                     self.adaptive_routers[model_name] = [
                         *self.adaptive_routers.get(model_name, []),
-                        TaggedPreRoutingStrategy(tags=tagged.tags, strategy=adaptive_router),
+                        TaggedPreRoutingStrategy(
+                            tags=tagged.tags,
+                            strategy=adaptive_router,
+                            definition_fingerprint=tagged.definition_fingerprint,
+                            deployment=tagged.deployment,
+                        ),
                     ]
 
         self._sync_adaptive_router_hooks()
@@ -14224,6 +14264,7 @@ class Router:
         spend metadata is stamped before routing and the response carries the tier group the
         strategy picked.
         """
+        self._record_routing_decision(request_kwargs=request_kwargs, routing_decision=None)
         requested_registered_model_name: Final = self._get_model_from_alias(model=model) or model
         registered_model_name: Final = await self._resolve_claude_code_session_router(
             model=model,
@@ -14247,7 +14288,6 @@ class Router:
         )
         if selected_strategy is None:
             await arm_compaction(request_kwargs, None)
-            self._record_routing_decision(request_kwargs=request_kwargs, routing_decision=None)
             self._stamp_or_clear_metadata_key(
                 request_kwargs=request_kwargs, key=SESSION_DEPLOYMENT_AFFINITY_TTL_METADATA_KEY, value=None
             )
@@ -14259,6 +14299,26 @@ class Router:
         from litellm.proxy.auth.auto_router_checks import authorize_member_auto_router_inference
         from litellm.router_strategy.complexity_router.complexity_router import ComplexityRouter
 
+        marker: Final = self._selected_strategy_marker_deployment(
+            model=registered_model_name, strategy_tags=selected_strategy.tags, request_kwargs=request_kwargs
+        )
+        definition: Final = selected_strategy.deployment
+        config_id: Final = definition.model_info.id if definition else None
+        updated_at: Final = definition.model_info.updated_at if definition else None
+        router_kind: Final = classify_strategy_router_model(definition.litellm_params.model) if definition else None
+        provenance: Final[StandardLoggingRoutingDecision] = {
+            "router_model_name": definition.model_name if definition else registered_model_name,
+            **({"router_type": router_kind} if router_kind is not None else {}),
+            **({"router_config_id": config_id} if config_id is not None else {}),
+            **({"router_config_updated_at": updated_at.isoformat()} if updated_at is not None else {}),
+            **(
+                {"router_config_fingerprint": selected_strategy.definition_fingerprint}
+                if selected_strategy.definition_fingerprint is not None
+                else {}
+            ),
+        }
+        phase_attributes(routing_decision_attributes(provenance))
+
         reject_recursive_compactor(registered_model_name)
         await arm_compaction(
             request_kwargs,
@@ -14282,11 +14342,7 @@ class Router:
         )
 
         await authorize_member_auto_router_inference(
-            deployment=self._selected_strategy_marker_deployment(
-                model=registered_model_name,
-                strategy_tags=selected_strategy.tags,
-                request_kwargs=request_kwargs,
-            ),
+            deployment=marker,
             request_kwargs=request_kwargs,
             llm_router=self,
         )
@@ -14324,13 +14380,18 @@ class Router:
             input=input,
             specific_deployment=specific_deployment,
         )
+        traced_routed: Final = (
+            routed.model_copy(update={"routing_decision": {**provenance, **routed.routing_decision}})
+            if routed is not None and routed.routing_decision is not None
+            else routed
+        )
         # Routing-only compression must not leak into the response: the model call and
         # deployment-context filtering key off this field. Compared by value, since
         # pydantic rebuilds the list rather than keeping the object passed in.
         pre_routing_hook_response: Final = (
-            routed.model_copy(update={"messages": messages})
-            if routed is not None and routing_messages is not None and routed.messages == routing_messages
-            else routed
+            traced_routed.model_copy(update={"messages": messages})
+            if traced_routed is not None and routing_messages is not None and traced_routed.messages == routing_messages
+            else traced_routed
         )
         self._record_routing_decision(
             request_kwargs=request_kwargs,
@@ -14481,6 +14542,7 @@ class Router:
         """
         from litellm.types.router import BaselineRouteStamp
 
+        phase_attributes(routing_decision_attributes(routing_decision))
         baseline_model: Final = routing_decision.get("savings_baseline_model") if routing_decision else None
         baseline_id: Final = routing_decision.get("savings_baseline_deployment_id") if routing_decision else None
         router_name: Final = routing_decision.get("router_model_name") if routing_decision else None

litellm/router_strategy/auto_router/auto_router.py

@@ -4,12 +4,13 @@ Auto-Routing Strategy that works with a Semantic Router Config
 
 import asyncio
 from collections.abc import Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Final, Optional
+from typing import TYPE_CHECKING, Any, Final, NamedTuple, Optional
 
 from pydantic import ConfigDict
 
 from litellm._logging import verbose_router_logger
 from litellm.constants import DEFAULT_AUTO_ROUTER_MAX_INPUT_CHARS
+from litellm.exceptions import Timeout as LiteLLMTimeout
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.internal_call_metadata import (
     effective_turn_off_message_logging,
@@ -17,7 +18,12 @@ from litellm.litellm_core_utils.internal_call_metadata import (
     parent_session_kwargs,
 )
 from litellm.types.llms.base import LiteLLMBaseModel
-from litellm.types.utils import AUTOROUTER_CLASSIFIER_CALL_ORIGIN
+from litellm.types.utils import (
+    AUTOROUTER_CLASSIFIER_CALL_ORIGIN,
+    ClassifierFailureReason,
+    RoutingDecisionCause,
+    StandardLoggingRoutingDecision,
+)
 
 if TYPE_CHECKING:
     from semantic_router.routers import SemanticRouter
@@ -41,6 +47,12 @@ class _CallerMetadata(LiteLLMBaseModel):
     litellm_metadata: Mapping[str, object] | None = None
 
 
+class _SemanticMatchOutcome(NamedTuple):
+    route_name: str | None = None
+    failure_reason: ClassifierFailureReason | None = None
+    error_type: str | None = None
+
+
 class AutoRouter(CustomLogger):
     DEFAULT_AUTO_SYNC_VALUE = "local"
 
@@ -72,6 +84,7 @@ class AutoRouter(CustomLogger):
 
         from litellm.router_strategy.auto_router.litellm_encoder import LiteLLMRouterEncoder
 
+        self.model_name: Final = model_name
         self.auto_router_config_path: str | None = auto_router_config_path
         self.auto_router_config: str | None = auto_router_config
         self.auto_sync_value = self.DEFAULT_AUTO_SYNC_VALUE
@@ -205,20 +218,38 @@ class AutoRouter(CustomLogger):
         if resolved_messages is None:
             return None
 
-        routelayer = await self._ensure_routelayer()
+        routelayer: Final = await self._ensure_routelayer()
 
         message_content: Final = self._extract_text_from_messages(resolved_messages)
-        route_name: Final = await self._matched_route_name(routelayer, message_content, request_kwargs)
+        outcome: Final = await self._match_route(routelayer, message_content, request_kwargs)
+        routed_model: Final = outcome.route_name or self.default_model
+        cause: Final[RoutingDecisionCause] = (
+            "semantic_error"
+            if outcome.failure_reason is not None
+            else "semantic_match"
+            if outcome.route_name
+            else "semantic_no_match"
+        )
+        decision: Final[StandardLoggingRoutingDecision] = {
+            "router_model_name": self.model_name,
+            "router_type": "semantic",
+            "routed_model": routed_model,
+            "cause": cause,
+            "classifier_model": self.embedding_model,
+            **({"classifier_failure_reason": outcome.failure_reason} if outcome.failure_reason is not None else {}),
+            **({"classifier_error_type": outcome.error_type} if outcome.error_type is not None else {}),
+        }
 
         return PreRoutingHookResponse(
-            model=route_name or self.default_model,
+            model=routed_model,
             messages=messages,
+            routing_decision=decision,
         )
 
-    async def _matched_route_name(
+    async def _match_route(
         self, routelayer: "SemanticRouter", text: str, request_kwargs: Mapping[str, object]
-    ) -> str | None:
-        """Name of the route `text` matches, or None when nothing matched or the match failed.
+    ) -> _SemanticMatchOutcome:
+        """Matched route or the reason a default model will serve the request.
 
         `text` is embedded here rather than by `routelayer(text=...)` so the caller's metadata reaches
         `aembedding()` and the embedding's spend lands on the key/team that sent the request;
@@ -247,10 +278,16 @@ class AutoRouter(CustomLogger):
             verbose_router_logger.warning(
                 "AutoRouter: semantic routing failed (%s), falling back to default model %s", e, self.default_model
             )
-            return None
+            failure: Final = e.__cause__ if isinstance(e, ValueError) and e.__cause__ is not None else e
+            return _SemanticMatchOutcome(
+                failure_reason="timeout"
+                if isinstance(failure, (TimeoutError, asyncio.TimeoutError, LiteLLMTimeout))
+                else "classifier_error",
+                error_type=type(failure).__name__,
+            )
         verbose_router_logger.debug("route_choice: %s", route_choice)
         if isinstance(route_choice, RouteChoice):
-            return route_choice.name
+            return _SemanticMatchOutcome(route_name=route_choice.name)
         if isinstance(route_choice, list) and route_choice:
-            return route_choice[0].name
-        return None
+            return _SemanticMatchOutcome(route_name=route_choice[0].name)
+        return _SemanticMatchOutcome()

litellm/router_strategy/complexity_router/complexity_router.py

@@ -42,6 +42,7 @@ from litellm.constants import (
     SESSION_ID_GENERATED_METADATA_KEY,
 )
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.otel.runtime import phase_event
 from litellm.litellm_core_utils.classifier_logging import masked_originating_request
 from litellm.litellm_core_utils.core_helpers import (
     _get_parent_otel_span_from_kwargs,
@@ -77,6 +78,7 @@ from litellm.types.llms.openai import (
 )
 from litellm.types.utils import (
     AUTOROUTER_CLASSIFIER_CALL_ORIGIN,
+    ClassifierFailureReason,
     ModelResponse,
     RoutingDecisionCause,
     StandardLoggingHeuristicV2Forecast,
@@ -1085,10 +1087,21 @@ class ClassificationOutcome(NamedTuple):
     llm_v2_forecast: LLMV2Decision | None = None
     jev_verdict: JevVerdict | None = None
     heuristic_v2_forecast: StandardLoggingHeuristicV2Forecast | None = None
+    classifier_failure_reason: ClassifierFailureReason | None = None
+    classifier_error_type: str | None = None
 
 
-def _with_signal(outcome: ClassificationOutcome, signal: str | None) -> ClassificationOutcome:
-    return outcome if signal is None else outcome._replace(signals=(*outcome.signals, signal))
+def _with_classifier_failure(
+    outcome: ClassificationOutcome,
+    failure_reason: ClassifierFailureReason,
+    error_type: str | None,
+    signal: str | None,
+) -> ClassificationOutcome:
+    return outcome._replace(
+        signals=(*outcome.signals, signal) if signal is not None else outcome.signals,
+        classifier_failure_reason=failure_reason,
+        classifier_error_type=error_type,
+    )
 
 
 def _with_llm_v2_forecast(
@@ -1221,6 +1234,12 @@ def _is_classifier_timeout(exc: BaseException) -> bool:
     return isinstance(exc, LiteLLMTimeout)
 
 
+def _classifier_failure_reason(exc: Exception) -> ClassifierFailureReason:
+    if _is_classifier_timeout(exc):
+        return "timeout"
+    return "invalid_response" if isinstance(exc, ValidationError) else "classifier_error"
+
+
 def _allowed(models: tuple[str, ...], fit_filter: frozenset[str] | None) -> tuple[str, ...]:
     return models if fit_filter is None else tuple(model for model in models if model in fit_filter)
 
@@ -1835,6 +1854,8 @@ class ComplexityRouter(CustomLogger):
         escalated: bool = False,
         classifier_model: str | None = None,
         classifier_cost: float | None = None,
+        classifier_failure_reason: ClassifierFailureReason | None = None,
+        classifier_error_type: str | None = None,
         conversation_continuing: bool = True,
         tier_litellm_params: Mapping[str, object] | None = None,
         context_escalation_original_tier: ComplexityTier | str | None = None,
@@ -1853,6 +1874,12 @@ class ComplexityRouter(CustomLogger):
             routed_model=routed_model,
             cause=cause,
             conversation_continuing=conversation_continuing,
+            **(
+                {"classifier_failure_reason": classifier_failure_reason}
+                if classifier_failure_reason is not None
+                else {}
+            ),
+            **({"classifier_error_type": classifier_error_type} if classifier_error_type is not None else {}),
         )
         if (baseline := self.savings_baseline) is not None:
             decision["savings_baseline_model"] = baseline.model
@@ -2094,7 +2121,9 @@ class ComplexityRouter(CustomLogger):
         permit: Final = breaker.acquire_permit() if breaker is not None else None
         if breaker is not None and permit is None:
             return self._capability_classifier_failure_outcome(
-                "capability classifier circuit is open", signal=_CLASSIFIER_CIRCUIT_OPEN_SIGNAL
+                "capability classifier circuit is open",
+                failure_reason="circuit_open",
+                signal=_CLASSIFIER_CIRCUIT_OPEN_SIGNAL,
             )
         try:
             tier, classifier_cost, forecast = await self._classify_with_capability_llm(prompt, request_kwargs, messages)
@@ -2118,9 +2147,20 @@ class ComplexityRouter(CustomLogger):
         except Exception as e:  # noqa: BLE001 -- every unavailable or invalid judge verdict must fail closed
             if breaker is not None and permit is not None:
                 breaker.record_failure(permit, is_timeout=_is_classifier_timeout(e))
-            return self._capability_classifier_failure_outcome(f"capability classifier failed ({type(e).__name__})")
+            return self._capability_classifier_failure_outcome(
+                f"capability classifier failed ({type(e).__name__})",
+                failure_reason=_classifier_failure_reason(e),
+                error_type=type(e).__name__,
+            )
 
-    def _capability_classifier_failure_outcome(self, reason: str, signal: str | None = None) -> ClassificationOutcome:
+    def _capability_classifier_failure_outcome(
+        self,
+        reason: str,
+        *,
+        failure_reason: ClassifierFailureReason,
+        error_type: str | None = None,
+        signal: str | None = None,
+    ) -> ClassificationOutcome:
         """Fail closed to the configured capable tier without consulting another taxonomy."""
         capability: Final = self.config.capability_classifier_config
         if capability is None:
@@ -2141,6 +2181,8 @@ class ComplexityRouter(CustomLogger):
             score=None,
             signals=signals,
             cause="capability_classifier_fallback",
+            classifier_failure_reason=failure_reason,
+            classifier_error_type=error_type,
         )
 
     async def _llm_classifier_outcome(
@@ -2164,6 +2206,7 @@ class ComplexityRouter(CustomLogger):
                 prompt,
                 system_prompt,
                 scored,
+                failure_reason="circuit_open",
                 signal=_CLASSIFIER_CIRCUIT_OPEN_SIGNAL,
             )
         try:
@@ -2193,7 +2236,12 @@ class ComplexityRouter(CustomLogger):
             if breaker is not None and permit is not None:
                 breaker.record_failure(permit, is_timeout=_is_classifier_timeout(e))
             return self._classifier_failure_outcome(
-                f"LLM classifier failed ({type(e).__name__})", prompt, system_prompt, scored
+                f"LLM classifier failed ({type(e).__name__})",
+                prompt,
+                system_prompt,
+                scored,
+                failure_reason=_classifier_failure_reason(e),
+                error_type=type(e).__name__,
             )
 
     async def _jev_classifier_outcome(
@@ -2206,11 +2254,16 @@ class ComplexityRouter(CustomLogger):
         config: Final = self.config.opensource_classifier_config
         client: Final = self._jev_client
         if config is None or client is None:
-            return self._classifier_failure_outcome("jev classifier is not configured", prompt, system_prompt)
+            return self._classifier_failure_outcome(
+                "jev classifier is not configured", prompt, system_prompt, failure_reason="not_configured"
+            )
         marker_pairs: Final = self._reminder_markers_for_request(request_kwargs or EMPTY_MAPPING)
         if _encrypted_classifier_task(request_kwargs, marker_pairs) is not None:
             return self._classifier_failure_outcome(
-                "jev classifier does not support encrypted agent tasks", prompt, system_prompt
+                "jev classifier does not support encrypted agent tasks",
+                prompt,
+                system_prompt,
+                failure_reason="unsupported_input",
             )
         breaker: Final = self._classifier_circuit_breaker
         permit: Final = breaker.acquire_permit() if breaker is not None else None
@@ -2219,6 +2272,7 @@ class ComplexityRouter(CustomLogger):
                 "jev classifier circuit is open",
                 prompt,
                 system_prompt,
+                failure_reason="circuit_open",
                 signal=_CLASSIFIER_CIRCUIT_OPEN_SIGNAL,
             )
         criteria: Final[Mapping[str, str]] = (
@@ -2288,7 +2342,11 @@ class ComplexityRouter(CustomLogger):
             if breaker is not None and permit is not None:
                 breaker.record_failure(permit, is_timeout=_is_classifier_timeout(e))
             return self._classifier_failure_outcome(
-                f"jev classifier failed ({type(e).__name__})", prompt, system_prompt
+                f"jev classifier failed ({type(e).__name__})",
+                prompt,
+                system_prompt,
+                failure_reason=_classifier_failure_reason(e),
+                error_type=type(e).__name__,
             )
 
     def _classifier_failure_outcome(
@@ -2297,6 +2355,9 @@ class ComplexityRouter(CustomLogger):
         prompt: str,
         system_prompt: str | None,
         scored: ClassificationOutcome | None = None,
+        *,
+        failure_reason: ClassifierFailureReason,
+        error_type: str | None = None,
         signal: str | None = None,
     ) -> ClassificationOutcome:
         """The outcome when the LLM classifier or classifier plugin produced no usable tier:
@@ -2307,35 +2368,44 @@ class ComplexityRouter(CustomLogger):
         v2: Final = self.config.llm_v2_config
         if v2 is not None:
             verbose_router_logger.warning("ComplexityRouter: %s, routing to llm_v2 capable tier", reason)
-            return _with_signal(
+            return _with_classifier_failure(
                 ClassificationOutcome(
                     tier=ComplexityTier(v2.capable_tier),
                     score=None,
                     signals=("llm-v2:fallback-capable",),
                     cause="llm_v2_fallback",
                 ),
+                failure_reason,
+                error_type,
                 signal,
             )
         fallback_tier: Final = self.config.fallback_tier
         if fallback_tier is not None:
             verbose_router_logger.warning("ComplexityRouter: %s, routing to fallback_tier %s", reason, fallback_tier)
-            return _with_signal(
+            return _with_classifier_failure(
                 ClassificationOutcome(
                     tier=fallback_tier,
                     score=None,
                     signals=(f"classifier-fallback:{fallback_tier}",),
                     cause="classifier_fallback",
                 ),
+                failure_reason,
+                error_type,
                 signal,
             )
         verbose_router_logger.warning(
             "ComplexityRouter: %s, falling back to %s", reason, self.config.classifier_fallback
         )
         if self.config.classifier_fallback == "default_model":
-            return _with_signal(self._default_model_fallback_outcome(), signal)
+            return _with_classifier_failure(self._default_model_fallback_outcome(), failure_reason, error_type, signal)
         if scored is not None:
-            return _with_signal(scored, signal)
-        return _with_signal(self._classify_locally(prompt, system_prompt), signal)
+            return _with_classifier_failure(scored, failure_reason, error_type, signal)
+        return _with_classifier_failure(
+            self._classify_locally(prompt, system_prompt),
+            failure_reason,
+            error_type,
+            signal,
+        )
 
     async def _classify_with_plugin(
         self,
@@ -2349,7 +2419,9 @@ class ComplexityRouter(CustomLogger):
 
         plugin: Final = self.config.classifier_plugin
         if plugin is None:
-            return self._classifier_failure_outcome("classifier_plugin is not set", prompt, system_prompt)
+            return self._classifier_failure_outcome(
+                "classifier_plugin is not set", prompt, system_prompt, failure_reason="not_configured"
+            )
         kwargs: Final = request_kwargs if request_kwargs is not None else EMPTY_MAPPING
         pools: Final = self._tier_pools()
         try:
@@ -2365,29 +2437,48 @@ class ComplexityRouter(CustomLogger):
             verdict: Final = await asyncio.wait_for(
                 plugin.classify(context), timeout=self.config.classifier_plugin_timeout_ms / 1000
             )
-        except asyncio.TimeoutError:
+        except asyncio.TimeoutError as e:
             return self._classifier_failure_outcome(
-                f"classifier plugin timed out after {self.config.classifier_plugin_timeout_ms}ms", prompt, system_prompt
+                f"classifier plugin timed out after {self.config.classifier_plugin_timeout_ms}ms",
+                prompt,
+                system_prompt,
+                failure_reason="timeout",
+                error_type=type(e).__name__,
             )
         except Exception as e:  # noqa: BLE001 -- an operator hook can fail in arbitrary ways (network, bug); any failure must fall back rather than fail the request
-            return self._classifier_failure_outcome(f"classifier plugin failed ({e})", prompt, system_prompt)
+            return self._classifier_failure_outcome(
+                f"classifier plugin failed ({e})",
+                prompt,
+                system_prompt,
+                failure_reason=_classifier_failure_reason(e),
+                error_type=type(e).__name__,
+            )
         if verdict is None:
-            return self._classifier_failure_outcome("classifier plugin declined to classify", prompt, system_prompt)
+            return self._classifier_failure_outcome(
+                "classifier plugin declined to classify", prompt, system_prompt, failure_reason="declined"
+            )
         if not isinstance(verdict, str):
             return self._classifier_failure_outcome(
                 f"classifier plugin returned a non-string verdict of type {type(verdict).__name__}",
                 prompt,
                 system_prompt,
+                failure_reason="invalid_response",
             )
         tier: Final = self.config.resolve_classified_tier(verdict)
         if tier is None:
             return self._classifier_failure_outcome(
-                f"classifier plugin returned unknown tier {verdict!r}", prompt, system_prompt
+                f"classifier plugin returned unknown tier {verdict!r}",
+                prompt,
+                system_prompt,
+                failure_reason="invalid_response",
             )
         tier_key: Final = _tier_name(tier)
         if not pools.get(tier_key):
             return self._classifier_failure_outcome(
-                f"classifier plugin returned tier {tier_key!r}, which has no models configured", prompt, system_prompt
+                f"classifier plugin returned tier {tier_key!r}, which has no models configured",
+                prompt,
+                system_prompt,
+                failure_reason="invalid_response",
             )
         return ClassificationOutcome(
             tier=tier,
@@ -2651,9 +2742,13 @@ class ComplexityRouter(CustomLogger):
             verdict: Final = LLMV2Verdict.model_validate_json(extract_classifier_json(content))
         except ValidationError as error:
             _log_rejected_classifier_verdict(error, content, request_kwargs)
-            return self._classifier_failure_outcome("Invalid LLM V2 forecast", prompt, system_prompt)._replace(
-                classifier_cost=classifier_cost
-            )
+            return self._classifier_failure_outcome(
+                "Invalid LLM V2 forecast",
+                prompt,
+                system_prompt,
+                failure_reason="invalid_response",
+                error_type=type(error).__name__,
+            )._replace(classifier_cost=classifier_cost)
         decision: Final = v2.classify(verdict)
         return ClassificationOutcome(
             tier=ComplexityTier(v2.efficient_tier if decision.use_efficient else v2.capable_tier),
@@ -4094,6 +4189,10 @@ class ComplexityRouter(CustomLogger):
         try:
             semantic_tier: Final = await self._semantic_tier_override(user_message, request_kwargs)
         except Exception as e:  # noqa: BLE001 -- embedding call can fail many ways (timeout, provider/network/parse error); any failure must fall back to scoring, never fail the request
+            phase_event(
+                "litellm.router.semantic_keyword_fallback",
+                {"litellm.router.failure_reason": _classifier_failure_reason(e), "error.type": type(e).__name__},
+            )
             verbose_router_logger.warning(
                 "ComplexityRouter: semantic keyword matching failed (%s), falling back to complexity scoring", e
             )
@@ -4728,6 +4827,8 @@ class ComplexityRouter(CustomLogger):
                     conversation_continuing=conversation_continuing,
                     cause=outcome.cause,
                     signals=outcome.signals,
+                    classifier_failure_reason=outcome.classifier_failure_reason,
+                    classifier_error_type=outcome.classifier_error_type,
                     escalation_keyword=escalation_keyword,
                     escalated=False,
                     tier_litellm_params=fallback_tier_params,
@@ -4840,6 +4941,8 @@ class ComplexityRouter(CustomLogger):
             escalated=escalated,
             classifier_model=classifier_model,
             classifier_cost=outcome.classifier_cost,
+            classifier_failure_reason=outcome.classifier_failure_reason,
+            classifier_error_type=outcome.classifier_error_type,
             tier_litellm_params=tier_litellm_params,
             context_escalation_original_tier=context_original_tier,
         )

litellm/types/router.py

@@ -1214,6 +1214,8 @@ class TaggedPreRoutingStrategy(Generic[_PreRoutingStrategyT_co]):
 
     tags: tuple[str, ...]
     strategy: _PreRoutingStrategyT_co
+    definition_fingerprint: str | None = None
+    deployment: Deployment | None = None
 
 
 @dataclass(frozen=True, slots=True)

litellm/types/utils.py

@@ -3061,6 +3061,14 @@ RoutingDecisionCause = Literal[
     "keyword",
     "quality_tier",
     "bandit",
+    "semantic_match",
+    "semantic_no_match",
+    "semantic_error",
+]
+
+
+ClassifierFailureReason = Literal[
+    "timeout", "circuit_open", "not_configured", "unsupported_input", "invalid_response", "declined", "classifier_error"
 ]
 
 
@@ -3093,7 +3101,10 @@ class StandardLoggingRoutingDecision(TypedDict, total=False):
     """Per-request provenance for a pre-routing strategy (auto-router) decision."""
 
     router_model_name: str
-    router_type: Literal["complexity", "adaptive", "quality"]
+    router_type: ReadOnly[Literal["complexity", "adaptive", "quality", "semantic"]]
+    router_config_id: ReadOnly[str]
+    router_config_updated_at: ReadOnly[str]
+    router_config_fingerprint: ReadOnly[str]
     routed_model: str
     cause: RoutingDecisionCause
     tier: str
@@ -3105,6 +3116,8 @@ class StandardLoggingRoutingDecision(TypedDict, total=False):
     escalation_keyword: str
     classifier_model: str
     classifier_cost: float
+    classifier_failure_reason: ReadOnly[ClassifierFailureReason]
+    classifier_error_type: ReadOnly[str]
     classifier_probabilities: ReadOnly[Mapping[str, float]]
     classifier_confidence: ReadOnly[float]
     heuristic_v2_forecast: ReadOnly[StandardLoggingHeuristicV2Forecast]
@@ -3143,6 +3156,9 @@ DERIVED_ROUTING_DECISION_FIELDS: Final[frozenset[str]] = frozenset(
     {
         "router_model_name",
         "router_type",
+        "router_config_id",
+        "router_config_updated_at",
+        "router_config_fingerprint",
         "routed_model",
         "cause",
         "tier",
@@ -3151,6 +3167,8 @@ DERIVED_ROUTING_DECISION_FIELDS: Final[frozenset[str]] = frozenset(
         "score",
         "classifier_model",
         "classifier_cost",
+        "classifier_failure_reason",
+        "classifier_error_type",
         "classifier_probabilities",
         "classifier_confidence",
         "heuristic_v2_forecast",

tests/unit/integrations/otel/test_otel_v2_logger.py

@@ -182,10 +182,63 @@ def test_async_log_success_event_emits_llm_call_span():
     assert span.attributes[GenAI.OPERATION_NAME] == "chat"
     assert span.attributes[GenAI.REQUEST_MODEL] == "gpt-4o"
     assert span.attributes[LiteLLM.CALL_ID] == "call_1"
+    assert not any(key.startswith("litellm.routing.") for key in span.attributes)
     # Success leaves status UNSET (semconv default), not forced OK.
     assert span.status.status_code is StatusCode.UNSET
 
 
+@pytest.mark.parametrize("fail", [False, True])
+@pytest.mark.parametrize("legacy_compat", [False, True])
+def test_routing_diagnostics_reach_llm_spans_without_promoting_prompt_text(fail: bool, legacy_compat: bool) -> None:
+    logger, exporter = _logger(legacy_compat=legacy_compat)
+    decision: Final = {
+        "router_model_name": "auto-router",
+        "router_type": "complexity",
+        "router_config_id": "definition-1",
+        "router_config_updated_at": "2026-10-06T00:00:00Z",
+        "router_config_fingerprint": "definition-hash",
+        "routed_model": "answer",
+        "cause": "heuristic",
+        "score": 0.25,
+        "classifier_cost": 0.001,
+        "classifier_failure_reason": "timeout",
+        "classifier_error_type": "TimeoutError",
+        "escalated": False,
+    }
+    payload: Final = _payload(
+        metadata={
+            "routing_decision": {
+                **decision,
+                "signals": ["private prompt"],
+                "matched_keyword": "private prompt",
+                "escalation_keyword": "private prompt",
+                "classifier_crux": "private prompt",
+                "tier_litellm_params": {"api_key": "private key"},
+                "classifier_probabilities": {"easy": 0.25},
+                "unknown_field": "private config",
+                "tier_label": {"invalid": "scalar field"},
+            }
+        },
+        status="failure" if fail else "success",
+    )
+    _emit_llm(logger, _kwargs(payload), fail=fail)
+    (span,) = exporter.get_finished_spans()
+    actual: Final = {key: value for key, value in span.attributes.items() if key.startswith("litellm.routing.")}
+    assert actual == {f"litellm.routing.{key}": value for key, value in decision.items()}
+    assert isinstance(actual["litellm.routing.score"], float)
+    assert actual["litellm.routing.escalated"] is False
+
+
+@pytest.mark.parametrize("decision", ["custom", 42, ["custom"]])
+def test_malformed_routing_metadata_preserves_the_completed_llm_span(decision: object) -> None:
+    logger, exporter = _logger()
+    _emit_llm(logger, _kwargs(_payload(metadata={"routing_decision": decision})))
+    (span,) = exporter.get_finished_spans()
+    assert span.attributes[GenAI.REQUEST_MODEL] == "gpt-4o"
+    assert span.attributes[LiteLLM.CALL_ID] == "call_1"
+    assert not any(key.startswith("litellm.routing.") for key in span.attributes)
+
+
 def test_llm_call_span_carries_the_callers_conversation_id():
     logger, exporter = _logger()
     kwargs = {**_kwargs(), "litellm_params": {"litellm_session_id": "conv-42", "metadata": {}}}
@@ -1103,6 +1156,29 @@ def test_phase_event_lands_on_root_span_even_inside_active_phase_span():
     assert by_name["auth /chat/completions"].events == ()
 
 
+def test_routing_attributes_stay_on_the_active_phase_while_retry_events_stay_on_root() -> None:
+    logger, exporter = _logger()
+    root: Final = logger.tracer.start_span("request", kind=SpanKind.SERVER)
+    set_request_root_span(root)
+    with trace.use_span(root, end_on_exit=True):
+        logger.set_phase_attributes({"litellm.routing.cause": "outside phase"})
+        with logger.start_phase_span("route auto-router"):
+            logger.set_phase_attributes({"litellm.routing.router_config_id": "definition-1"})
+            with logger.tracer.start_as_current_span("classifier transport"):
+                logger.set_phase_attributes({"litellm.routing.cause": "heuristic"})
+                logger.add_phase_event("litellm.request.retry", {"error.type": "TimeoutError"})
+        logger.set_phase_attributes({"litellm.routing.cause": "after phase"})
+    by_name: Final = {span.name: span for span in exporter.get_finished_spans()}
+    route: Final = by_name["route auto-router"]
+    assert route.attributes["litellm.routing.router_config_id"] == "definition-1"
+    assert route.attributes["litellm.routing.cause"] == "heuristic"
+    assert route.parent.span_id == root.get_span_context().span_id
+    assert route.events == ()
+    assert "litellm.routing.cause" not in by_name["request"].attributes
+    assert "litellm.routing.cause" not in by_name["classifier transport"].attributes
+    assert by_name["request"].events[0].attributes["error.type"] == "TimeoutError"
+
+
 def test_live_llm_span_anchors_to_root_with_no_active_span():
     """Bug 2 (pass-through), live path: even with no span active at ``pre_call``,
     the anchor is a recordable parent, so the span opens live under the server root

tests/unit/integrations/otel/test_runtime.py

@@ -10,6 +10,11 @@ import lock. These tests pin the import to a single resolution.
 import builtins
 import importlib.abc
 import sys
+from collections.abc import Sequence
+from types import ModuleType
+from typing import Final
+
+import pytest
 
 import litellm.integrations.otel.runtime as runtime
 
@@ -92,4 +97,28 @@ def test_phase_span_does_not_import_the_proxy_in_an_sdk_process(monkeypatch):
     with runtime.phase_span("route gpt-5-mini") as span:
         assert span is None
 
+    assert runtime.phase_attributes({"litellm.routing.score": 0.25}) is None
     assert proxy_imports == []
+
+
+def test_phase_attributes_no_op_when_sdk_import_is_unavailable(monkeypatch: pytest.MonkeyPatch) -> None:
+    runtime._otel_runtime.cache_clear()
+    real_import: Final = builtins.__import__
+
+    def without_sdk(
+        name: str,
+        globals: dict[str, object] | None = None,
+        locals: dict[str, object] | None = None,
+        fromlist: Sequence[str] = (),
+        level: int = 0,
+    ) -> ModuleType:
+        if name == "litellm.integrations.otel" and "logger" in fromlist:
+            raise ImportError("OpenTelemetry SDK is not installed")
+        return real_import(name, globals, locals, fromlist, level)
+
+    monkeypatch.setattr(builtins, "__import__", without_sdk)
+    try:
+        assert runtime.phase_attributes({"litellm.routing.score": 0.25}) is None
+        assert runtime._otel_runtime() is None
+    finally:
+        runtime._otel_runtime.cache_clear()

tests/unit/integrations/test_opentelemetry.py

@@ -310,6 +310,65 @@ class TestOpenTelemetryTeamAttributesOnChildSpans(unittest.TestCase):
         )
 
 
+def test_v1_routing_diagnostics_add_scalar_attributes_and_preserve_legacy_metadata() -> None:
+    exporter: Final = InMemorySpanExporter()
+    provider: Final = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    otel: Final = OpenTelemetry(tracer_provider=provider)
+    decision: Final = {
+        "router_model_name": "auto-router",
+        "routed_model": "answer",
+        "router_config_id": "definition-1",
+        "cause": "heuristic",
+        "classifier_failure_reason": "timeout",
+        "classifier_error_type": "TimeoutError",
+        "classifier_cost": 0.001,
+    }
+    with provider.get_tracer(__name__).start_as_current_span("inference") as span:
+        otel.set_attributes(
+            span,
+            {
+                "model": "answer",
+                "standard_logging_object": {
+                    "id": "routing-call",
+                    "call_type": "acompletion",
+                    "metadata": {"routing_decision": decision},
+                },
+            },
+            {"model": "answer", "choices": [], "usage": {}},
+        )
+    (finished,) = exporter.get_finished_spans()
+    actual: Final = {
+        key: value for key, value in finished.attributes.items() if key.startswith("litellm.routing.")
+    }
+    assert actual == {f"litellm.routing.{key}": value for key, value in decision.items()}
+    assert finished.attributes["metadata.routing_decision"] == str(decision)
+
+
+@pytest.mark.parametrize("decision", ["custom", 42, ["custom"]])
+def test_v1_malformed_routing_metadata_preserves_inference_attributes(decision: object) -> None:
+    exporter: Final = InMemorySpanExporter()
+    provider: Final = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    otel: Final = OpenTelemetry(tracer_provider=provider)
+    with provider.get_tracer(__name__).start_as_current_span("inference") as span:
+        otel.set_attributes(
+            span,
+            {
+                "model": "answer",
+                "standard_logging_object": {
+                    "id": "routing-call",
+                    "call_type": "acompletion",
+                    "metadata": {"routing_decision": decision},
+                },
+            },
+            {"model": "answer", "choices": [], "usage": {}},
+        )
+    (finished,) = exporter.get_finished_spans()
+    assert finished.attributes["gen_ai.request.model"] == "answer"
+    assert not any(key.startswith("litellm.routing.") for key in finished.attributes)
+
+
 class TestOpenTelemetryCostBreakdown(unittest.TestCase):
     def test_cost_breakdown_emitted_to_otel_span(self):
         """

tests/unit/proxy/management_endpoints/test_auto_router_endpoints.py

@@ -1222,15 +1222,14 @@ class TestAutoRouterBenchmarks:
         assert [group.router_name for group in response.groups] == ["tagged"]
 
     def test_the_listed_kinds_match_the_router_types_traffic_can_record(self):
-        """The one reason semantic is excluded, pinned against both declarations: a kind the
-        rollup can record must be listable, and a kind it cannot must not be."""
         from typing import get_args, get_type_hints
 
         from litellm.router_utils.auto_router_model_naming import StrategyRouterKind
         from litellm.types.utils import StandardLoggingRoutingDecision
 
-        recorded = set(get_args(get_type_hints(StandardLoggingRoutingDecision)["router_type"]))
-        assert set(get_args(StrategyRouterKind)) - {"semantic"} == recorded
+        readonly_router_type: Final = get_type_hints(StandardLoggingRoutingDecision, include_extras=True)["router_type"]
+        recorded: Final = set(get_args(get_args(readonly_router_type)[0]))
+        assert set(get_args(StrategyRouterKind)) == recorded
 
 
 # ---------------------------------------------------------------------------

tests/unit/router_strategy/test_auto_router.py

@@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+from litellm.exceptions import Timeout as LiteLLMTimeout
 from litellm.router_strategy.auto_router.auto_router import AutoRouter
 
 pytestmark_skip_beta = pytest.mark.skip(
@@ -342,6 +343,8 @@ class FixedRouteLayer:
         self.route_choice = route_choice
 
     async def acall(self, vector: Any) -> Any:
+        if isinstance(self.route_choice, BaseException):
+            raise self.route_choice
         return self.route_choice
 
 
@@ -372,11 +375,15 @@ class StubEmbeddingRouter:
 class FailingEmbeddingRouter(StubEmbeddingRouter):
     """Router whose query embedding fails, as it does when the prompt exceeds the encoder's window."""
 
-    async def aembedding(self, input: List[str], model: str, **kwargs: Any) -> Any:
-        raise ValueError(
+    def __init__(self, error: Exception | None = None) -> None:
+        super().__init__()
+        self.error = error or ValueError(
             "litellm.InternalServerError: input is too large to process. increase the physical batch size"
         )
 
+    async def aembedding(self, input: List[str], model: str, **kwargs: Any) -> Any:
+        raise self.error
+
 
 def _auto_router(routelayer: Any, litellm_router_instance: Any = None, **kwargs: Any) -> AutoRouter:
     auto_router: Final = AutoRouter(
@@ -395,24 +402,35 @@ class TestAutoRouterAlwaysResolvesARoutableModel:
     """The hook returns the alias's default model instead of failing or leaking the alias downstream."""
 
     @pytest.mark.asyncio
-    async def test_should_fall_back_to_default_model_when_the_embedding_call_fails(self):
+    @pytest.mark.parametrize("request_model", ["my-auto-router", "public-router-alias"])
+    async def test_should_fall_back_to_default_model_when_the_embedding_call_fails(self, request_model: str) -> None:
         auto_router: Final = _auto_router(FixedRouteLayer(None), litellm_router_instance=FailingEmbeddingRouter())
 
         result: Final = await auto_router.async_pre_routing_hook(
-            model="my-auto-router",
+            model=request_model,
             request_kwargs={},
             messages=[{"role": "user", "content": "a" * 100_000}],
         )
 
         assert result is not None
         assert result.model == "fallback-model"
+        assert result.routing_decision == {
+            "router_model_name": "my-auto-router",
+            "router_type": "semantic",
+            "routed_model": "fallback-model",
+            "cause": "semantic_error",
+            "classifier_model": auto_router.embedding_model,
+            "classifier_failure_reason": "classifier_error",
+            "classifier_error_type": "ValueError",
+        }
 
     @pytest.mark.asyncio
-    async def test_should_fall_back_to_default_model_when_no_route_matches(self):
+    @pytest.mark.parametrize("request_model", ["my-auto-router", "public-router-alias"])
+    async def test_should_fall_back_to_default_model_when_no_route_matches(self, request_model: str) -> None:
         auto_router: Final = _auto_router(FixedRouteLayer(None))
 
         result: Final = await auto_router.async_pre_routing_hook(
-            model="my-auto-router",
+            model=request_model,
             request_kwargs={},
             messages=[{"role": "user", "content": "nothing like any route"}],
         )
@@ -420,6 +438,13 @@ class TestAutoRouterAlwaysResolvesARoutableModel:
         assert result is not None
         # Leaving "my-auto-router" here fails downstream with "Unmapped LLM provider".
         assert result.model == "fallback-model"
+        assert result.routing_decision == {
+            "router_model_name": "my-auto-router",
+            "router_type": "semantic",
+            "routed_model": "fallback-model",
+            "cause": "semantic_no_match",
+            "classifier_model": auto_router.embedding_model,
+        }
 
     @pytest.mark.asyncio
     async def test_should_fall_back_to_default_model_when_the_route_layer_returns_an_empty_list(self):
@@ -452,14 +477,15 @@ class TestAutoRouterAlwaysResolvesARoutableModel:
         assert result.model == "code-model"
 
     @pytest.mark.asyncio
-    async def test_should_still_route_to_the_matched_route_when_one_matches(self):
+    @pytest.mark.parametrize("request_model", ["my-auto-router", "public-router-alias"])
+    async def test_should_still_route_to_the_matched_route_when_one_matches(self, request_model: str) -> None:
         from semantic_router.schema import RouteChoice
 
         router: Final = StubEmbeddingRouter()
         auto_router: Final = _auto_router(FixedRouteLayer(RouteChoice(name="code-model")), litellm_router_instance=router)
 
         result: Final = await auto_router.async_pre_routing_hook(
-            model="my-auto-router",
+            model=request_model,
             request_kwargs={},
             messages=[{"role": "user", "content": "fix this stack trace"}],
         )
@@ -467,6 +493,43 @@ class TestAutoRouterAlwaysResolvesARoutableModel:
         assert result is not None
         assert result.model == "code-model"
         assert router.seen_text == "fix this stack trace"
+        assert result.routing_decision == {
+            "router_model_name": "my-auto-router",
+            "router_type": "semantic",
+            "routed_model": "code-model",
+            "cause": "semantic_match",
+            "classifier_model": auto_router.embedding_model,
+        }
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("error", (
+        TimeoutError("private request content"),
+        asyncio.TimeoutError("private request content"),
+        LiteLLMTimeout(message="private request content", model="embedding", llm_provider="openai"),
+    ))
+    @pytest.mark.parametrize("wrapped", [False, True])
+    async def test_should_record_timeout_without_exception_message(self, error: Exception, wrapped: bool) -> None:
+        auto_router: Final = _auto_router(
+            FixedRouteLayer(None if wrapped else error),
+            litellm_router_instance=FailingEmbeddingRouter(error) if wrapped else StubEmbeddingRouter(),
+        )
+        result: Final = await auto_router.async_pre_routing_hook(
+            model="my-auto-router", request_kwargs={}, messages=[{"role": "user", "content": "route this"}]
+        )
+        assert result is not None and result.model == "fallback-model"
+        assert result.routing_decision is not None
+        assert result.routing_decision["cause"] == "semantic_error"
+        assert result.routing_decision["classifier_failure_reason"] == "timeout"
+        assert result.routing_decision["classifier_error_type"] == type(error).__name__
+        assert "private request content" not in str(result.routing_decision)
+
+    @pytest.mark.asyncio
+    async def test_should_propagate_cancellation_during_matching(self) -> None:
+        auto_router: Final = _auto_router(FixedRouteLayer(asyncio.CancelledError()))
+        with pytest.raises(asyncio.CancelledError):
+            await auto_router.async_pre_routing_hook(
+                model="my-auto-router", request_kwargs={}, messages=[{"role": "user", "content": "route this"}]
+            )
 
 
 class TestAutoRouterEmbeddingInputCap:

tests/unit/router_strategy/test_complexity_router.py

@@ -20,11 +20,12 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import httpx
 import pytest
 import respx
-from pydantic import ValidationError
+from pydantic import TypeAdapter, ValidationError
 
 import litellm
 from litellm import Router
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.types.utils import StandardLoggingRoutingDecision
 from litellm.router_utils.auto_router_model_naming import (
     CUSTOMIZATION_CAPABILITY,
     GATED_AUTO_ROUTER_CAPABILITIES,
@@ -484,6 +485,10 @@ class TestComplexityRouterInit:
         assert first.cause != "jev_classifier"
         assert second.cause != "jev_classifier"
         assert client.calls == 1
+        assert first.classifier_failure_reason == "timeout"
+        assert first.classifier_error_type == "TimeoutError"
+        assert second.classifier_failure_reason == "circuit_open"
+        assert second.classifier_error_type is None
         assert _CLASSIFIER_CIRCUIT_OPEN_SIGNAL in second.signals
 
     @pytest.mark.asyncio
@@ -518,6 +523,8 @@ class TestComplexityRouterInit:
         outcome = await router.aclassify("Explain this")
 
         assert outcome.cause != "jev_classifier"
+        assert outcome.classifier_failure_reason == "classifier_error"
+        assert outcome.classifier_error_type == ("RuntimeError" if isinstance(response, RuntimeError) else "ValueError")
 
 
 class TestTokenScoring:
@@ -3342,6 +3349,8 @@ class TestCapabilityClassifier:
         assert outcome.tier == ComplexityTier.REASONING
         assert outcome.cause == "capability_classifier_fallback"
         assert outcome.signals == ("capability-classifier-fallback",)
+        assert outcome.classifier_failure_reason == "invalid_response"
+        assert outcome.classifier_error_type == "ValidationError"
 
     @pytest.mark.asyncio
     async def test_classifier_call_failure_fails_closed_to_capable_model(self, mock_router_instance):
@@ -3353,6 +3362,8 @@ class TestCapabilityClassifier:
         )
         assert response.model == "capable-model"
         assert response.routing_decision["cause"] == "capability_classifier_fallback"
+        assert response.routing_decision["classifier_failure_reason"] == "timeout"
+        assert response.routing_decision["classifier_error_type"] == "TimeoutError"
         assert "classifier_p_solve" not in response.routing_decision
         assert "classifier_threshold" not in response.routing_decision
 
@@ -4226,6 +4237,9 @@ class TestLLMClassifier:
         assert first.cause == "heuristic_scorer"
         assert second.cause == "heuristic_scorer"
         assert "classifier-circuit-open" in second.signals
+        assert first.classifier_failure_reason == "timeout"
+        assert second.classifier_failure_reason == "circuit_open"
+        assert second.classifier_error_type is None
         mock_router_instance.acompletion.assert_awaited_once()
 
     def test_classifier_circuit_allows_one_probe_and_closes_on_success(self):
@@ -8189,18 +8203,24 @@ class TestClassifierPlugin:
         router = _plugin_router(mock_router_instance, _FixedTierClassifier(None))
         outcome = await router.aclassify("what is 2+2?")
         assert outcome.cause == "heuristic_scorer"
+        assert outcome.classifier_failure_reason == "declined"
+        assert outcome.classifier_error_type is None
 
     @pytest.mark.asyncio
     async def test_plugin_error_falls_back_to_heuristic(self, mock_router_instance):
         router = _plugin_router(mock_router_instance, _RaisingClassifier())
         outcome = await router.aclassify("what is 2+2?")
         assert outcome.cause == "heuristic_scorer"
+        assert outcome.classifier_failure_reason == "classifier_error"
+        assert outcome.classifier_error_type == "RuntimeError"
 
     @pytest.mark.asyncio
     async def test_plugin_timeout_falls_back_to_heuristic(self, mock_router_instance):
         router = _plugin_router(mock_router_instance, _SlowClassifier(), classifier_plugin_timeout_ms=20)
         outcome = await router.aclassify("what is 2+2?")
         assert outcome.cause == "heuristic_scorer"
+        assert outcome.classifier_failure_reason == "timeout"
+        assert outcome.classifier_error_type == "TimeoutError"
 
     @pytest.mark.asyncio
     async def test_plugin_non_string_verdict_falls_back_to_heuristic(self, mock_router_instance):
@@ -8208,12 +8228,16 @@ class TestClassifierPlugin:
         router = _plugin_router(mock_router_instance, _FixedTierClassifier(42))
         outcome = await router.aclassify("what is 2+2?")
         assert outcome.cause == "heuristic_scorer"
+        assert outcome.classifier_failure_reason == "invalid_response"
+        assert outcome.classifier_error_type is None
 
     @pytest.mark.asyncio
     async def test_plugin_unknown_tier_falls_back_to_heuristic(self, mock_router_instance):
         router = _plugin_router(mock_router_instance, _FixedTierClassifier("galactic"))
         outcome = await router.aclassify("what is 2+2?")
         assert outcome.cause == "heuristic_scorer"
+        assert outcome.classifier_failure_reason == "invalid_response"
+        assert outcome.classifier_error_type is None
 
     @pytest.mark.asyncio
     async def test_plugin_tier_without_pool_falls_back(self, mock_router_instance):
@@ -8229,6 +8253,8 @@ class TestClassifierPlugin:
         )
         outcome = await router.aclassify("what is 2+2?")
         assert outcome.cause == "heuristic_scorer"
+        assert outcome.classifier_failure_reason == "invalid_response"
+        assert outcome.classifier_error_type is None
 
     @pytest.mark.asyncio
     async def test_plugin_failure_with_default_model_fallback(self, mock_router_instance):
@@ -9564,6 +9590,101 @@ class TestRedactedLoggingDropsPromptText:
         assert decision["cause"] == "literal_keyword_match"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "overrides, expected_cause",
+    [
+        ({"classifier_type": "llm"}, "heuristic_scorer"),
+        ({"classifier_type": "llm", "classifier_fallback": "default_model"}, "default_model_fallback"),
+        (
+            {
+                "classifier_type": "capability",
+                "capability_classifier_config": {
+                    "efficient_tier": "SIMPLE", "capable_tier": "REASONING",
+                    "base_threshold": 0.5, "threshold_step": 0.1,
+                },
+            },
+            "capability_classifier_fallback",
+        ),
+        (
+            {
+                "classifier_type": "llm_v2",
+                "llm_v2_config": {
+                    "efficient_profile": "Routine tasks", "capable_profile": "Complex tasks",
+                    "harness": "Read files and run checks", "max_quality_gap": 0.05,
+                },
+            },
+            "llm_v2_fallback",
+        ),
+    ],
+)
+async def test_classifier_failure_survives_real_router_decision_and_redaction(
+    overrides: Mapping[str, object], expected_cause: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router: Final = Router(model_list=[
+        {
+            "model_name": "auto",
+            "litellm_params": {
+                "model": "auto_router/complexity_router",
+                "complexity_router_config": {
+                    "tiers": {"SIMPLE": "cheap", "REASONING": "capable"},
+                    "default_model": "cheap",
+                    "classifier_llm_config": {"model": "classifier", "timeout_ms": 2000},
+                    **overrides,
+                },
+            },
+        },
+        {
+            "model_name": "classifier",
+            "litellm_params": {
+                "model": "openai/test-classifier", "api_key": "test-key",
+                "api_base": "https://classifier.test/v1",
+            },
+        },
+        {"model_name": "cheap", "litellm_params": {"model": "openai/test-cheap", "api_key": "test-key"}},
+        {"model_name": "capable", "litellm_params": {"model": "openai/test-capable", "api_key": "test-key"}},
+    ])
+    with respx.mock:
+        upstream: Final = respx.post("https://classifier.test/v1/chat/completions").respond(
+            200,
+            json={
+                "id": "classifier-result", "object": "chat.completion", "model": "test-classifier",
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": "PRIVATE_INVALID_REPLY"},
+                             "finish_reason": "stop"}],
+            },
+        )
+        metadata: Final[dict[str, object]] = {"headers": {"x-litellm-enable-message-redaction": True}}
+        request: Final[dict[str, object]] = {"metadata": metadata, "turn_off_message_logging": True}
+        response: Final = await router.async_pre_routing_hook(
+            model="auto", request_kwargs=request, messages=[{"role": "user", "content": "hi"}]
+        )
+        assert upstream.called
+        assert response is not None and response.routing_decision is not None
+        decision: Final = TypeAdapter(StandardLoggingRoutingDecision).validate_python(metadata["routing_decision"])
+        assert decision["cause"] == expected_cause
+        assert decision["classifier_failure_reason"] == "invalid_response"
+        assert decision["classifier_error_type"] == "ValidationError"
+        assert "PRIVATE_INVALID_REPLY" not in json.dumps(decision)
+        assert "signals" not in decision
+        if overrides["classifier_type"] == "llm":
+            upstream.respond(200, json={
+                "id": "classifier-recovered", "object": "chat.completion", "model": "test-classifier",
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": '{"tier":"SIMPLE"}'},
+                             "finish_reason": "stop"}],
+            })
+            recovered: Final = await router.async_pre_routing_hook(
+                model="auto", request_kwargs=request, messages=[{"role": "user", "content": "hi"}]
+            )
+            assert recovered is not None and recovered.routing_decision is not None
+            assert recovered.routing_decision["cause"] == "llm_classifier"
+            recovered_decision: Final = TypeAdapter(StandardLoggingRoutingDecision).validate_python(
+                metadata["routing_decision"]
+            )
+            assert "classifier_failure_reason" not in recovered_decision
+            assert "classifier_error_type" not in recovered_decision
+
+
 def test_every_routing_decision_field_is_classified():
     """Redaction is derived from a declaration, not a list at the call site, so every
     field has to be classified as quoting the prompt or aggregating it. A field added
@@ -14148,6 +14269,8 @@ async def test_v2_chain_judge_failure_uses_selected_fallback(
     assert outcome.cause == ("heuristic_v2" if fallback == "heuristic" else "default_model_fallback")
     assert outcome.tier == (ComplexityTier.COMPLEX if fallback == "heuristic" else ComplexityTier.MEDIUM)
     assert (outcome.heuristic_v2_forecast is not None) == (fallback == "heuristic")
+    assert outcome.classifier_failure_reason == "classifier_error"
+    assert outcome.classifier_error_type == "RuntimeError"
     assert dependency.aresponses.await_count == int(encrypted)
     assert dependency.acompletion.await_count == int(not encrypted)
 

tests/unit/test_router/test_router.py

@@ -12,7 +12,7 @@ import warnings
 from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Mapping
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
-from typing import Final, Literal
+from typing import TYPE_CHECKING, Final, Literal
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
@@ -31,6 +31,7 @@ from litellm.exceptions import GuardrailRaisedException, MidStreamFallbackError,
 from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.otel.plumbing.context import set_request_root_span
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
 from litellm.llms.anthropic.pass_through.messages.agentic_streaming_iterator import (
     SERVER_FULFILLED_TOOL_LEAK_ERROR_SSE_BYTES,
@@ -68,8 +69,14 @@ from litellm.types.router import (
     ModelInfo,
     PreRoutingHookResponse,
     RetryPolicy,
+    RoutingContext,
 )
 
+if TYPE_CHECKING:
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+
+    from litellm.integrations.otel.logger import OpenTelemetryV2
+
 
 def test_update_kwargs_does_not_mutate_defaults_and_merges_metadata():
     # initialize a real Router (env‑vars can be empty)
@@ -13213,15 +13220,22 @@ class TestTeamPublicNameReachesPreRoutingStrategies:
 
     @pytest.mark.asyncio
     async def test_team_public_name_shadows_a_global_model_for_that_team_only(self):
-        router = self._router(
-            {self.INTERNAL_NAME: self._RewriteStrategy()},
-            extra_deployments=({"model_name": self.PUBLIC_NAME, "litellm_params": {"model": "openai/gpt-4o"}},),
+        router: Final = litellm.Router(
+            model_list=[
+                self._team_marker(self.INTERNAL_NAME),
+                {"model_name": "gemini-flash", "litellm_params": {"model": "gemini/gemini-3.6-flash"}},
+                {"model_name": self.PUBLIC_NAME, "litellm_params": {"model": "openai/gpt-4o"}},
+            ],
         )
 
         async def routed(request_kwargs: dict) -> str | None:
             response = await router.async_pre_routing_hook(
                 model=self.PUBLIC_NAME, request_kwargs=request_kwargs, messages=self._messages()
             )
+            if response is not None:
+                assert response.routing_decision is not None
+                assert response.routing_decision["router_model_name"] == self.INTERNAL_NAME
+                assert response.routing_decision["router_config_id"] == router.model_list[0]["model_info"]["id"]
             return response.model if response else None
 
         async def selected(request_kwargs: dict) -> str:
@@ -20267,3 +20281,270 @@ async def test_non_chat_surfaces_mark_their_deployment_pick(monkeypatch: pytest.
     router.completion(model="gpt-4o", messages=[{"role": "user", "content": "hi"}])
 
     assert events == [_pick("embed", "initial", 1), _pick("gpt-4o", "initial", 1)]
+
+
+class TestAutoRouterTraceProvenance:
+    @pytest.fixture
+    def _tracing(self, monkeypatch: pytest.MonkeyPatch) -> "tuple[OpenTelemetryV2, InMemorySpanExporter]":
+        from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+        from litellm.integrations.otel import OpenTelemetryV2Config
+        from litellm.integrations.otel.logger import OpenTelemetryV2
+        from litellm.integrations.otel.plumbing import providers
+        from litellm.proxy import proxy_server
+
+        config: Final = OpenTelemetryV2Config(exporter="in_memory")
+        exporter: Final = InMemorySpanExporter()
+        logger: Final = OpenTelemetryV2(config=config, tracer_provider=providers.build_tracer_provider(config, exporter=exporter))
+        monkeypatch.setattr(proxy_server, "open_telemetry_logger", logger)
+        return logger, exporter
+
+    @staticmethod
+    def _marker(tag: str, target: str = "answer") -> dict[str, object]:
+        return {
+            "model_name": "traced-router",
+            "model_info": {"id": f"definition-{tag}", "updated_at": "2026-10-01T12:00:00+00:00"},
+            "litellm_params": {
+                "model": "auto_router/complexity_router", "tags": [tag],
+                "complexity_router_config": {
+                    "tiers": dict.fromkeys(("SIMPLE", "MEDIUM", "COMPLEX", "REASONING"), target),
+                },
+            },
+        }
+
+    def test_definition_fingerprint_is_canonical_and_tracks_config_not_credentials(self):
+        original: Final = Deployment.model_validate(self._marker("blue"))
+        reordered: Final = Deployment.model_validate({
+            **self._marker("blue"),
+            "litellm_params": {
+                "api_key": "credential-only-change", "tags": ["blue"], "model": "auto_router/complexity_router",
+                "complexity_router_config": {"tiers": dict.fromkeys(("REASONING", "COMPLEX", "MEDIUM", "SIMPLE"), "answer")},
+            },
+        })
+        fingerprint: Final = Router._routing_definition_fingerprint(original)
+        assert fingerprint is not None
+        assert fingerprint == Router._routing_definition_fingerprint(reordered)
+        assert fingerprint != Router._routing_definition_fingerprint(Deployment.model_validate(self._marker("blue", "other")))
+
+    @pytest.mark.asyncio
+    async def test_direct_classifier_and_plugin_prelude_never_inherit_parent_decisions(
+        self, _tracing: "tuple[OpenTelemetryV2, InMemorySpanExporter]", monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        logger, exporter = _tracing
+        completed: Final = asyncio.Event()
+
+        class ObservedClassifier:
+            prelude: Mapping[str, object] = {}
+
+            async def run(self, context: RoutingContext) -> RoutingContext:
+                self.prelude = dict(context.metadata)
+                return context
+
+            async def classify(self, context: RoutingContext) -> str:
+                await litellm.acompletion(
+                    model="openai/gpt-5.6-luna", api_key="test", mock_response="SIMPLE",
+                    messages=[{"role": "user", "content": "choose a tier"}],
+                    metadata={**context.metadata, "internal_call_origin": "autorouter_classifier"},
+                )
+                return "SIMPLE"
+
+        class CallbackComplete(CustomLogger):
+            async def async_log_success_event(
+                self, kwargs: Mapping[str, object], response_obj: object, start_time: datetime, end_time: datetime
+            ) -> None:
+                completed.set()
+
+        monkeypatch.setattr(litellm, "_async_success_callback", [logger, CallbackComplete()])
+        classifier: Final = ObservedClassifier()
+        router: Final = Router(model_list=[
+            {**self._marker("blue"), "litellm_params": {
+                "model": "auto_router/complexity_router",
+                "complexity_router_config": {
+                    "classifier_type": "custom", "classifier_plugin": classifier,
+                    "tiers": {"SIMPLE": "answer"}, "session_affinity": False, "deployment_affinity": False,
+                },
+            }},
+            {"model_name": "answer", "litellm_params": {"model": "openai/gpt-5.6-luna", "api_key": "test"}},
+        ], plugins=[classifier])
+        stale: Final = {"routing_decision": {"router_model_name": "previous"}, "_autorouter_baseline_route": "old"}
+        request: Final = {"metadata": {**stale}, "litellm_metadata": {**stale, "user_api_key_team_id": "team"}}
+        with logger.tracer.start_as_current_span("request") as root:
+            set_request_root_span(root)
+            await router.async_get_available_deployment(
+                model="traced-router", request_kwargs=request, messages=[{"role": "user", "content": "hello"}]
+            )
+            await completed.wait()
+        classifier_span: Final = next(
+            span for span in exporter.get_finished_spans()
+            if (span.attributes or {}).get("litellm.request.purpose") == "autorouter_classifier"
+        )
+        assert not any(key.startswith("litellm.routing.") for key in classifier_span.attributes)
+        assert "routing_decision" not in classifier.prelude
+        assert "_autorouter_baseline_route" not in classifier.prelude
+        assert classifier.prelude["user_api_key_team_id"] == "team"
+        assert "routing_decision" not in request["metadata"]
+        assert "_autorouter_baseline_route" not in request["metadata"]
+        assert request["litellm_metadata"]["routing_decision"]["cause"] == "classifier_plugin"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("build_fails", [False, True])
+    async def test_semantic_internal_failure_keeps_phase_identity_without_parent_retry_attribution(
+        self, _tracing: "tuple[OpenTelemetryV2, InMemorySpanExporter]", monkeypatch: pytest.MonkeyPatch,
+        respx_mock: respx.MockRouter, build_fails: bool,
+    ) -> None:
+        logger, exporter = _tracing
+        monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+        query_statuses: Final = iter((503, 200))
+
+        def embedding_response(request: httpx.Request) -> httpx.Response:
+            status: Final = 400 if build_fails else (
+                next(query_statuses) if json.loads(request.content)["input"] == ["query-retry"] else 200
+            )
+            return httpx.Response(status, headers={"x-should-retry": "false"}, json={
+                "object": "list", "model": "text-embedding-3-small",
+                "data": [{"object": "embedding", "index": 0, "embedding": [1.0, 0.0]}],
+                "usage": {"prompt_tokens": 1, "total_tokens": 1},
+            })
+
+        upstream: Final = respx_mock.post("https://embedding.example/v1/embeddings").mock(side_effect=embedding_response)
+        router: Final = Router(model_list=[
+            {"model_name": "semantic", "model_info": {"id": "semantic-definition"}, "litellm_params": {
+                "model": "auto_router/semantic-router", "auto_router_default_model": "answer",
+                "auto_router_embedding_model": "embed",
+                "auto_router_config": json.dumps({"routes": [{"name": "answer", "utterances": ["hello"]}]}),
+            }},
+            *({"model_name": "embed", "model_info": {"id": f"embed-{index}"}, "litellm_params": {
+                "model": "openai/text-embedding-3-small", "api_base": "https://embedding.example/v1",
+                "api_key": "test", "max_retries": 0,
+            }} for index in range(2)),
+            {"model_name": "answer", "litellm_params": {"model": "openai/gpt-5.6-luna", "api_key": "test"}},
+        ], num_retries=1, retry_after=0)
+        stale: Final = {"routing_decision": {"router_model_name": "previous"}, "_autorouter_baseline_route": "old"}
+        request: Final = {"metadata": {**stale}, "litellm_metadata": {**stale}}
+        with logger.tracer.start_as_current_span("request") as root:
+            set_request_root_span(root)
+            call: Final = router.async_get_available_deployment(
+                model="semantic", request_kwargs=request, messages=[{"role": "user", "content": "query-retry"}]
+            )
+            if build_fails:
+                with pytest.raises(ValueError, match="API call failed"):
+                    await call
+            else:
+                await call
+        spans: Final = {span.name: span for span in exporter.get_finished_spans()}
+        assert spans["route semantic"].attributes["litellm.routing.router_config_id"] == "semantic-definition"
+        assert not any(event.name == "litellm.routing.retry" for event in spans["request"].events)
+        if build_fails:
+            assert all(
+                "routing_decision" not in bucket and "_autorouter_baseline_route" not in bucket
+                for bucket in (request["metadata"], request["litellm_metadata"])
+            )
+        else:
+            assert sum(json.loads(call.request.content)["input"] == ["query-retry"] for call in upstream.calls) == 2
+            assert request["litellm_metadata"]["routing_decision"]["cause"] == "semantic_match"
+            assert sum(span.name == "route embed" for span in exporter.get_finished_spans()) == 2
+            assert any(event.name == "litellm.request.deployment_selected" for event in spans["request"].events)
+
+    @pytest.mark.asyncio
+    async def test_same_alias_and_tags_across_strategy_families_keep_selected_definition(
+        self, _tracing: "tuple[OpenTelemetryV2, InMemorySpanExporter]"
+    ) -> None:
+        logger, exporter = _tracing
+        router: Final = Router(model_list=[
+            self._marker("blue"),
+            {
+                "model_name": "traced-router",
+                "model_info": {"id": "definition-semantic"},
+                "litellm_params": {
+                    "model": "auto_router/semantic-router",
+                    "tags": ["blue"],
+                    "auto_router_config": json.dumps({"routes": [{"name": "answer", "utterances": ["hello"]}]}),
+                    "auto_router_default_model": "answer",
+                    "auto_router_embedding_model": "embed",
+                },
+            },
+        ])
+        with logger.start_phase_span("route traced-router"):
+            response: Final = await router.async_pre_routing_hook(
+                model="traced-router", request_kwargs={"metadata": {"tags": ["blue"]}}
+            )
+        assert response is None
+        span: Final = next(span for span in exporter.get_finished_spans() if span.name == "route traced-router")
+        assert span.attributes is not None
+        assert span.attributes["litellm.routing.router_config_id"] == router.model_list[1]["model_info"]["id"]
+        assert span.attributes["litellm.routing.router_type"] == "semantic"
+
+    @pytest.mark.asyncio
+    async def test_semantic_team_alias_keeps_registered_identity_after_routing_and_retry(
+        self,
+        _tracing: "tuple[OpenTelemetryV2, InMemorySpanExporter]",
+        monkeypatch: pytest.MonkeyPatch,
+        respx_mock: respx.MockRouter,
+    ) -> None:
+        logger, exporter = _tracing
+        monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+        respx_mock.post("https://embedding.example/v1/embeddings").respond(200, json={
+            "object": "list",
+            "model": "text-embedding-3-small",
+            "data": [{"object": "embedding", "index": 0, "embedding": [1.0, 0.0]}],
+            "usage": {"prompt_tokens": 1, "total_tokens": 1},
+        })
+        router: Final = Router(model_list=[
+            {
+                "model_name": "internal-semantic-router",
+                "model_info": {"id": "semantic-definition", "team_id": "trace-team", "team_public_model_name": "public-router"},
+                "litellm_params": {
+                    "model": "auto_router/semantic-router",
+                    "auto_router_config": json.dumps({"routes": [{"name": "answer", "utterances": ["hello"]}]}),
+                    "auto_router_default_model": "answer",
+                    "auto_router_embedding_model": "embed",
+                },
+            },
+            {"model_name": "embed", "litellm_params": {"model": "openai/text-embedding-3-small", "api_base": "https://embedding.example/v1", "api_key": "test"}},
+            {"model_name": "answer", "litellm_params": {"model": "openai/gpt-5.6-luna", "api_key": "test"}},
+        ])
+        request: Final = {"metadata": {"user_api_key_team_id": "trace-team"}}
+        with logger.tracer.start_as_current_span("request"):
+            await router.async_get_available_deployment(
+                model="public-router", request_kwargs=request, messages=[{"role": "user", "content": "hello"}]
+            )
+            router.log_retry({**request, "model": "answer"}, ValueError("retry"))
+        decision: Final = request["metadata"]["routing_decision"]
+        assert decision["router_model_name"] == router.model_list[0]["model_name"]
+        assert decision["router_config_id"] == router.model_list[0]["model_info"]["id"]
+        assert decision["cause"] == "semantic_match"
+        spans: Final = {span.name: span for span in exporter.get_finished_spans()}
+        assert spans["route public-router"].attributes["litellm.routing.router_model_name"] == decision["router_model_name"]
+        assert spans["route public-router"].attributes["litellm.routing.router_config_id"] == decision["router_config_id"]
+        retry: Final = next(event for event in spans["request"].events if event.name == "litellm.routing.retry")
+        assert retry.attributes["litellm.routing.router_model_name"] == decision["router_model_name"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("passthrough", [False, True])
+    async def test_attempt_spans_keep_selected_config_and_plain_fallback_clears_it(
+        self, _tracing: "tuple[OpenTelemetryV2, InMemorySpanExporter]", passthrough: bool
+    ) -> None:
+        logger, exporter = _tracing
+        router: Final = Router(model_list=[
+            self._marker("blue"), self._marker("green"),
+            {"model_name": "answer", "litellm_params": {"model": "openai/gpt-5.6-luna", "api_key": "test", "use_in_pass_through": True, "tags": ["blue", "green"]}},
+        ], enable_tag_filtering=True)
+        request: Final = {"metadata": {"tags": ["green"]}, "litellm_metadata": {"tags": ["green"]}}
+        pick: Final = router.async_get_available_deployment_for_pass_through if passthrough else router.async_get_available_deployment
+        with logger.tracer.start_as_current_span("request"):
+            await pick(model="traced-router", request_kwargs=request, messages=[{"role": "user", "content": "hello"}])
+            decision: Final = request["litellm_metadata"]["routing_decision"]
+            assert decision["router_config_id"] == "definition-green"
+            assert decision["router_config_updated_at"] == "2026-10-01T12:00:00+00:00"
+            assert decision["router_config_fingerprint"] == router.complexity_routers["traced-router"][1].definition_fingerprint
+            assert "routing_decision" not in request["metadata"]
+            router.log_retry({**request, "model": "answer"}, ValueError("private-error-text"))
+            await pick(model="answer", request_kwargs=request, messages=[{"role": "user", "content": "hello"}])
+        assert "routing_decision" not in request["litellm_metadata"]
+        spans: Final = {span.name: span for span in exporter.get_finished_spans()}
+        assert spans["route traced-router"].attributes["litellm.routing.router_config_id"] == "definition-green"
+        assert spans["route traced-router"].attributes["litellm.routing.routed_model"] == "answer"
+        assert not any(key.startswith("litellm.routing.") for key in spans["route answer"].attributes)
+        retry: Final = next(event for event in spans["request"].events if event.name == "litellm.routing.retry")
+        assert retry.attributes["error.type"] == "ValueError"
+        assert retry.attributes["litellm.retry.count"] == 1
+        assert "private-error-text" not in str(retry.attributes)

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

@@ -45475,7 +45475,7 @@ export interface components {
              * Cause
              * @enum {string}
              */
-            cause?: "prompt_cache_cost" | "heuristic_scorer" | "heuristic_v2" | "reasoning_override" | "llm_classifier" | "capability_classifier" | "jev_classifier" | "llm_v2_classifier" | "llm_v2_fallback" | "heuristic_first_short_circuit" | "hybrid_short_circuit" | "classifier_plugin" | "classifier_fallback" | "capability_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";
+            cause?: "prompt_cache_cost" | "heuristic_scorer" | "heuristic_v2" | "reasoning_override" | "llm_classifier" | "capability_classifier" | "jev_classifier" | "llm_v2_classifier" | "llm_v2_fallback" | "heuristic_first_short_circuit" | "hybrid_short_circuit" | "classifier_plugin" | "classifier_fallback" | "capability_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" | "semantic_match" | "semantic_no_match" | "semantic_error";
             /** Classifier Calibrated Capable P Solve */
             classifier_calibrated_capable_p_solve?: number;
             /** Classifier Calibrated Efficient P Solve */
@@ -45496,6 +45496,13 @@ export interface components {
             classifier_crux?: string;
             /** Classifier Efficient P Solve */
             classifier_efficient_p_solve?: number;
+            /** Classifier Error Type */
+            classifier_error_type?: string;
+            /**
+             * Classifier Failure Reason
+             * @enum {string}
+             */
+            classifier_failure_reason?: "timeout" | "circuit_open" | "not_configured" | "unsupported_input" | "invalid_response" | "declined" | "classifier_error";
             /** Classifier Max Quality Gap */
             classifier_max_quality_gap?: number;
             /** Classifier Model */
@@ -45531,13 +45538,19 @@ export interface components {
             request_type?: string;
             /** Routed Model */
             routed_model?: string;
+            /** Router Config Fingerprint */
+            router_config_fingerprint?: string;
+            /** Router Config Id */
+            router_config_id?: string;
+            /** Router Config Updated At */
+            router_config_updated_at?: string;
             /** Router Model Name */
             router_model_name?: string;
             /**
              * Router Type
              * @enum {string}
              */
-            router_type?: "complexity" | "adaptive" | "quality";
+            router_type?: "complexity" | "adaptive" | "quality" | "semantic";
             /** Savings Baseline Deployment Id */
             savings_baseline_deployment_id?: string;
             /** Savings Baseline Model */