diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py
index 45c7825344b..66e36dab2ba 100644
--- a/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py
+++ b/litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py
@@ -86,22 +86,41 @@ def _decoded_sse_data_line(line: bytes) -> object | None:
         return None
 
 
-def _anthropic_error_event_payload(chunk: object) -> Mapping[str, object] | None:
+def _anthropic_event_payload(chunk: object, event_type: str) -> Mapping[str, object] | None:
     if isinstance(chunk, dict):
-        return chunk if chunk.get("type") == "error" else None
+        return chunk if chunk.get("type") == event_type else None
     if isinstance(chunk, (bytes, bytearray)):
         decoded_lines: Final = (_decoded_sse_data_line(line) for line in chunk.splitlines())
         return next(
             (
                 candidate
                 for candidate in decoded_lines
-                if isinstance(candidate, dict) and candidate.get("type") == "error"
+                if isinstance(candidate, dict) and candidate.get("type") == event_type
             ),
             None,
         )
     return None
 
 
+def _anthropic_error_event_payload(chunk: object) -> Mapping[str, object] | None:
+    return _anthropic_event_payload(chunk, "error")
+
+
+def parse_anthropic_refusal_stop_details(chunk: object) -> Mapping[str, object] | None:
+    """
+    Return the ``stop_details`` object of an Anthropic SSE ``message_delta``
+    chunk whose delta carries ``stop_reason: "refusal"`` (a safeguard refusal:
+    https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback),
+    or None for any other chunk, a plain refusal without ``stop_details`` included.
+    """
+    payload: Final = _anthropic_event_payload(chunk, "message_delta")
+    delta: Final = payload.get("delta") if payload is not None else None
+    if not isinstance(delta, dict) or delta.get("stop_reason") != "refusal":
+        return None
+    stop_details: Final = delta.get("stop_details")
+    return stop_details if isinstance(stop_details, dict) else None
+
+
 def _anthropic_error_body(chunk: object) -> Mapping[str, object] | None:
     """Return the ``error`` object of an Anthropic SSE ``event: error`` chunk, or None."""
     payload: Final = _anthropic_error_event_payload(chunk)
diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/utils.py b/litellm/llms/anthropic/experimental_pass_through/messages/utils.py
index 02d82887dde..9deff950724 100644
--- a/litellm/llms/anthropic/experimental_pass_through/messages/utils.py
+++ b/litellm/llms/anthropic/experimental_pass_through/messages/utils.py
@@ -1,11 +1,40 @@
+from collections.abc import Mapping
 from functools import lru_cache
-from typing import Any, Final, cast, get_type_hints
+from typing import TYPE_CHECKING, Any, Final, cast, get_type_hints
 
 from litellm.types.llms.anthropic import AnthropicMessagesRequestOptionalParams
 from litellm.types.llms.anthropic_messages.anthropic_response import (
     AnthropicMessagesResponse,
 )
 
+if TYPE_CHECKING:
+    from litellm.exceptions import ContentPolicyViolationError
+
+
+def get_safeguard_refusal_stop_details(response: object) -> Mapping[str, Any] | None:
+    """
+    Return the ``stop_details`` of an Anthropic Messages response refused by a
+    safeguard (``stop_reason: "refusal"`` carrying ``stop_details``:
+    https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback),
+    or None for any other response, a plain refusal without ``stop_details`` included.
+    """
+    if not isinstance(response, dict) or response.get("stop_reason") != "refusal":
+        return None
+    stop_details: Final = response.get("stop_details")
+    return stop_details if isinstance(stop_details, dict) else None
+
+
+def safeguard_refusal_error(model: str, stop_details: Mapping[str, object]) -> "ContentPolicyViolationError":
+    """The exception a safeguard-refused Anthropic response converts into so the
+    content-policy fallback chain can re-dispatch it."""
+    from litellm.exceptions import ContentPolicyViolationError
+
+    return ContentPolicyViolationError(
+        message=f"Anthropic safeguard refusal (category: {stop_details.get('category')}).",
+        model=model,
+        llm_provider="anthropic",
+    )
+
 
 @lru_cache(maxsize=1)
 def _anthropic_messages_optional_param_keys() -> frozenset[str]:
@@ -100,14 +129,12 @@ def mock_response(
             model=model,
         )
     return AnthropicMessagesResponse(
-        **{
-            "content": [{"text": mock_response, "type": "text"}],
-            "id": "msg_013Zva2CMHLNnXjNJJKqJ2EF",
-            "model": "claude-sonnet-4-20250514",
-            "role": "assistant",
-            "stop_reason": "end_turn",
-            "stop_sequence": None,
-            "type": "message",
-            "usage": {"input_tokens": 2095, "output_tokens": 503},
-        }
+        content=[{"text": mock_response, "type": "text"}],
+        id="msg_013Zva2CMHLNnXjNJJKqJ2EF",
+        model="claude-sonnet-4-20250514",
+        role="assistant",
+        stop_reason="end_turn",
+        stop_sequence=None,
+        type="message",
+        usage={"input_tokens": 2095, "output_tokens": 503},
     )
diff --git a/litellm/router.py b/litellm/router.py
index 471a1116f44..6e4405ebfef 100644
--- a/litellm/router.py
+++ b/litellm/router.py
@@ -143,7 +143,11 @@ from litellm.router_utils.cooldown_handlers import (
 from litellm.router_utils.fallback_event_handlers import (
     AttemptedFallbackTargets,
     _check_non_standard_fallback_format,
-    get_fallback_model_group,
+    clear_pre_routing_selection,
+    fallback_lookup_groups,
+    get_fallback_model_group_for_lookup_groups,
+    get_pre_routing_selection,
+    record_pre_routing_selection,
     run_async_fallback,
 )
 from litellm.router_utils.get_retry_from_policy import (
@@ -4918,6 +4922,19 @@ class Router:
                 )
                 response = await response
 
+            if self._should_raise_anthropic_refusal_error(
+                model=model,
+                original_generic_function=original_generic_function,
+                response=response,
+                kwargs=kwargs,
+            ):
+                from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
+                    safeguard_refusal_error,
+                )
+
+                refusal_details: Final = cast(dict, response["stop_details"])  # cast-ok: gate verified the shape
+                raise safeguard_refusal_error(model=model, stop_details=refusal_details)
+
             self.success_calls[model_name] += 1
             verbose_router_logger.info("ageneric_api_call_with_fallbacks(model=%s)\x1b[32m 200 OK\x1b[0m", model_name)
 
@@ -4964,6 +4981,11 @@ class Router:
         # fallback to the original reference for any non-picklable value.
         # The original_generic_function is preserved so the per-attempt
         # helper knows which underlying API to call on fallback.
+        # The pre-routing hook stamps its tier selection into this bucket during the primary
+        # attempt; seeding it before the snapshot gives both the live kwargs and the copy a
+        # bucket, so the post-call carry-over below always has somewhere to read and write.
+        kwargs.setdefault("litellm_metadata", {})  # mutable-ok: shared bucket  # rebind-ok: stamp must be readable here
+
         fallback_kwargs: Final[dict[str, object]] = kwargs.copy()
         if isinstance(fallback_kwargs.get("litellm_metadata"), dict):
             fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"])
@@ -4973,6 +4995,14 @@ class Router:
 
         response: Final = await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs)
 
+        # The snapshot predates the pre-routing hook, so the tier it stamped into the live kwargs
+        # is carried over write-or-clear: a stale or caller-supplied selection left in the copy
+        # would key the mid-stream fallback lookup off a tier this attempt never routed to.
+        clear_pre_routing_selection(fallback_kwargs)
+        live_pre_routing_selection: Final = get_pre_routing_selection(kwargs)
+        if live_pre_routing_selection is not None:
+            record_pre_routing_selection(fallback_kwargs, live_pre_routing_selection)
+
         if kwargs.get("stream") and isinstance(response, BaseResponsesAPIStreamingIterator):
             return await self._aresponses_streaming_iterator(
                 response=response,
@@ -5030,6 +5060,10 @@ class Router:
         from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
             aclose_if_supported,
             parse_anthropic_error_event,
+            parse_anthropic_refusal_stop_details,
+        )
+        from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
+            safeguard_refusal_error,
         )
 
         source_iterator: Final = response
@@ -5068,13 +5102,35 @@ class Router:
                         continue
                     if _anthropic_stream_commits_now(chunk, has_generated_content, len(buffered_lifecycle_chunks)):
                         has_generated_content = True  # rebind-ok: real content seen, or the buffer cap was hit
-                    error_event = parse_anthropic_error_event(chunk)
+                    # A transport can split one SSE data line across byte chunks, so pre-content
+                    # detection parses the accumulated buffer plus the current chunk, never the
+                    # chunk alone; the buffer is already capped, which bounds this window too.
+                    parse_window = (  # rebind-ok: freshly computed each iteration, never carried over
+                        b"".join(c for c in (*buffered_lifecycle_chunks, chunk) if isinstance(c, (bytes, bytearray)))  # pyright: ignore[reportUnnecessaryIsInstance]  # bridge-path chunks are not always bytes at runtime
+                        if not has_generated_content and isinstance(chunk, (bytes, bytearray))  # pyright: ignore[reportUnnecessaryIsInstance]  # bridge-path chunks are not always bytes at runtime
+                        else chunk
+                    )
+                    error_event = parse_anthropic_error_event(parse_window)
                     retriable_pending_error = (  # rebind-ok: freshly computed each iteration, never carried over
                         not has_generated_content
                         and error_event is not None
                         and _is_retriable_anthropic_status(error_event[2])
                         and not _anthropic_stream_error_is_gateway_verdict(chunk)
                     )
+                    refusal_stop_details = (  # rebind-ok: freshly computed each iteration, never carried over
+                        parse_anthropic_refusal_stop_details(parse_window)
+                        if not has_generated_content and error_event is None
+                        else None
+                    )
+                    if refusal_stop_details is not None and self._has_content_policy_fallback(model, initial_kwargs):
+                        refusal_error = safeguard_refusal_error(model=model, stop_details=refusal_stop_details)
+                        raise MidStreamFallbackError(
+                            message=refusal_error.message,
+                            model=model,
+                            llm_provider="anthropic",
+                            original_exception=refusal_error,
+                            is_pre_first_chunk=True,
+                        )
                     if not has_generated_content and not retriable_pending_error and error_event is None:
                         buffered_lifecycle_chunks = (*buffered_lifecycle_chunks, chunk)
                         continue
@@ -5186,8 +5242,13 @@ class Router:
                 kwargs=initial_kwargs,
                 metadata_variable_name="litellm_metadata",
             )
+            # The content-policy dispatch branch matches on the trigger's own type, so a refusal's
+            # MidStreamFallbackError envelope is unwrapped here or the wrong fallback list is consulted.
+            fallback_trigger: Final[Exception] = (
+                e.original_exception if isinstance(e.original_exception, litellm.ContentPolicyViolationError) else e
+            )
             fallback_response = await self.async_function_with_fallbacks_common_utils(  # rebind-ok: set on success
-                e=e,
+                e=fallback_trigger,
                 disable_fallbacks=False,
                 fallbacks=fallbacks,
                 context_window_fallbacks=context_window_fallbacks,
@@ -5243,6 +5304,11 @@ class Router:
         # share, leaking primary-deployment metadata into the mid-stream
         # fallback request. safe_deep_copy avoids deep-copying the full
         # kwargs (which can hold non-deepcopyable logging handles/clients).
+        # The pre-routing hook stamps its tier selection into this bucket during the primary
+        # attempt; seeding it before the snapshot gives both the live kwargs and the copy a
+        # bucket, so the post-call carry-over below always has somewhere to read and write.
+        kwargs.setdefault("litellm_metadata", {})  # mutable-ok: shared bucket  # rebind-ok: stamp must be readable here
+
         fallback_kwargs: Final[dict[str, object]] = kwargs.copy()  # mutable-ok: mutated below before re-entry
         if isinstance(fallback_kwargs.get("litellm_metadata"), dict):
             fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"])
@@ -5252,6 +5318,14 @@ class Router:
 
         response: Final = await self._ageneric_api_call_with_fallbacks(original_function=original_function, **kwargs)
 
+        # The snapshot predates the pre-routing hook, so the tier it stamped into the live kwargs
+        # is carried over write-or-clear: a stale or caller-supplied selection left in the copy
+        # would key the mid-stream fallback lookup off a tier this attempt never routed to.
+        clear_pre_routing_selection(fallback_kwargs)
+        live_pre_routing_selection: Final = get_pre_routing_selection(kwargs)
+        if live_pre_routing_selection is not None:
+            record_pre_routing_selection(fallback_kwargs, live_pre_routing_selection)
+
         if kwargs.get("stream") and hasattr(response, "__aiter__"):
             return await self._aanthropic_messages_streaming_iterator(
                 response=cast("AsyncIterator[bytes]", response),  # cast-ok: stream=True always returns a byte iterator
@@ -6807,6 +6881,9 @@ class Router:
         original_exception: Final = e
         fallback_model_group = None
         original_model_group: Final[str | None] = kwargs.get("model")
+        # A pre-routing hook (complexity / auto / adaptive / quality routers) picks a tier
+        # behind the router name, and fallbacks are configured per tier, not per router.
+        lookup_groups: Final[tuple[str, ...]] = fallback_lookup_groups(kwargs, model_group)
         fallback_failure_exception_str = ""
 
         if disable_fallbacks is True or original_model_group is None:
@@ -6851,15 +6928,15 @@ class Router:
             ]
             # Get external fallbacks — handle both standard and non-standard formats
             external_fallback_group: list | None = None
-            if fallbacks is not None and model_group is not None:
+            if fallbacks is not None and lookup_groups:
                 if _check_non_standard_fallback_format(fallbacks=fallbacks):
                     # Non-standard formats (e.g. ["claude-3-haiku"] or
                     # [{"model": "...", "messages": [...]}]) are passed through directly
                     external_fallback_group = fallbacks
                 else:
-                    external_fallback_group, generic_idx = get_fallback_model_group(
+                    external_fallback_group, generic_idx = get_fallback_model_group_for_lookup_groups(
                         fallbacks=fallbacks,
-                        model_group=cast(str, model_group),
+                        lookup_groups=lookup_groups,
                     )
                     if external_fallback_group is None and generic_idx is not None:
                         external_fallback_group = fallbacks[generic_idx]["*"]
@@ -6917,9 +6994,9 @@ class Router:
             if isinstance(e, litellm.ContextWindowExceededError):
                 if context_window_fallbacks is not None:
                     context_window_fallback_model_group: Final[list[str] | None] = (
-                        self._get_fallback_model_group_from_fallbacks(
+                        self._get_fallback_model_group_for_lookup_groups(
                             fallbacks=context_window_fallbacks,
-                            model_group=model_group,
+                            lookup_groups=lookup_groups,
                         )
                     )
                     if context_window_fallback_model_group is None:
@@ -6950,9 +7027,9 @@ class Router:
             elif isinstance(e, litellm.ContentPolicyViolationError):
                 if content_policy_fallbacks is not None:
                     content_policy_fallback_model_group: Final[list[str] | None] = (
-                        self._get_fallback_model_group_from_fallbacks(
+                        self._get_fallback_model_group_for_lookup_groups(
                             fallbacks=content_policy_fallbacks,
-                            model_group=model_group,
+                            lookup_groups=lookup_groups,
                         )
                     )
                     if content_policy_fallback_model_group is None:
@@ -6979,14 +7056,14 @@ class Router:
 
                     if litellm.expose_router_debug_in_errors:
                         e.message += f"\n{error_message}"
-            if fallbacks is not None and model_group is not None:
+            if fallbacks is not None and lookup_groups:
                 verbose_router_logger.debug("inside model fallbacks: %s", mask_sensitive_structure(fallbacks))
                 (
                     fallback_model_group,
                     generic_fallback_idx,
-                ) = get_fallback_model_group(
+                ) = get_fallback_model_group_for_lookup_groups(
                     fallbacks=fallbacks,  # if fallbacks = [{"gpt-3.5-turbo": ["claude-3-haiku"]}]
-                    model_group=cast(str, model_group),
+                    lookup_groups=lookup_groups,
                 )
                 ## if none, check for generic fallback
                 if fallback_model_group is None and generic_fallback_idx is not None:
@@ -6995,12 +7072,12 @@ class Router:
                 if fallback_model_group is None:
                     masked_fallbacks: Final = mask_sensitive_structure(fallbacks)
                     verbose_router_logger.info(
-                        "No fallback model group found for original model_group=%s. Fallbacks=%s",
-                        model_group,
+                        "No fallback model group found for lookup_groups=%s. Fallbacks=%s",
+                        " -> ".join(lookup_groups),
                         masked_fallbacks,
                     )
                     if hasattr(original_exception, "message") and litellm.expose_router_debug_in_errors:
-                        original_exception.message += f"No fallback model group found for original model_group={model_group}. Fallbacks={masked_fallbacks}"
+                        original_exception.message += f"No fallback model group found for lookup_groups={' -> '.join(lookup_groups)}. Fallbacks={masked_fallbacks}"
                     raise original_exception
 
                 input_kwargs.update(
@@ -7046,6 +7123,7 @@ class Router:
         If it fails after num_retries, fall back to another model group
         """
         model_group: Final[str | None] = kwargs.get("model")
+        clear_pre_routing_selection(kwargs)  # pyright: ignore[reportUnknownArgumentType]  # **kwargs is untyped at this boundary
         if not isinstance(kwargs.get("attempted_targets"), AttemptedFallbackTargets):
             _fallback_metadata_key: Final = _get_router_metadata_variable_name(
                 function_name=getattr(kwargs.get("original_function"), "__name__", None)
@@ -7471,6 +7549,24 @@ class Router:
                 break
         return fallback_model_group
 
+    def _get_fallback_model_group_for_lookup_groups(
+        self,
+        fallbacks: list[dict[str, list[str]]],  # mutable-ok: mirrors the sibling resolver's contract
+        lookup_groups: tuple[str, ...],
+    ) -> list[str] | None:  # mutable-ok: mirrors the sibling resolver's contract
+        """First lookup group whose exact-key chain resolves (tier first, then requested group)."""
+        return next(
+            (
+                resolved
+                for resolved in (
+                    self._get_fallback_model_group_from_fallbacks(fallbacks=fallbacks, model_group=group)
+                    for group in lookup_groups
+                )
+                if resolved is not None
+            ),
+            None,
+        )
+
     def _get_first_default_fallback(self) -> str | None:
         """
         Returns the first model from the default_fallbacks list, if it exists.
@@ -7886,6 +7982,31 @@ class Router:
                     return True
         return False
 
+    def _has_content_policy_fallback(self, model_group: str, kwargs: Mapping[str, Any]) -> bool:
+        """
+        Whether a content-policy fallback would resolve for this request, keyed the same way
+        async_function_with_fallbacks_common_utils resolves it: the tier a pre-routing hook
+        selected wins over the requested group. Raising without this returning True would turn
+        a deliverable response into an error the fallback chain cannot recover from.
+        """
+        content_policy_fallbacks: Final = kwargs.get("content_policy_fallbacks", self.content_policy_fallbacks)
+        if content_policy_fallbacks is not None:
+            return (
+                self._get_fallback_model_group_for_lookup_groups(
+                    fallbacks=content_policy_fallbacks,
+                    lookup_groups=fallback_lookup_groups(kwargs, model_group),
+                )
+                is not None
+            )
+        if self._has_default_fallbacks():
+            return True
+        verbose_router_logger.debug(
+            "No content-policy fallback available. Returning original response. model=%s, content_policy_fallbacks=%s",
+            model_group,
+            content_policy_fallbacks,
+        )
+        return False
+
     def _should_raise_content_policy_error(self, model: str, response: ModelResponse, kwargs: dict) -> bool:
         """
         Determines if a content policy error should be raised.
@@ -7898,27 +8019,26 @@ class Router:
             if response.choices[0].finish_reason != "content_filter":
                 return False
 
-        content_policy_fallbacks: Final = kwargs.get("content_policy_fallbacks", self.content_policy_fallbacks)
+        return self._has_content_policy_fallback(model, kwargs)
 
-        ### ONLY RAISE ERROR IF CP FALLBACK AVAILABLE ###
-        if content_policy_fallbacks is not None:
-            fallback_model_group = None
-            for item in content_policy_fallbacks:  # [{"gpt-3.5-turbo": ["gpt-4"]}]
-                if list(item.keys())[0] == model:
-                    fallback_model_group = item[model]
-                    break
-
-            if fallback_model_group is not None:
-                return True
-        elif self._has_default_fallbacks():  # default fallbacks set
-            return True
-
-        verbose_router_logger.debug(
-            "Content Policy Error occurred. No available fallbacks. Returning original response. model=%s, content_policy_fallbacks=%s",
-            model,
-            content_policy_fallbacks,
+    def _should_raise_anthropic_refusal_error(
+        self, model: str, original_generic_function: Callable, response: object, kwargs: Mapping[str, Any]
+    ) -> bool:
+        """
+        The /v1/messages twin of _should_raise_content_policy_error: an Anthropic safeguard
+        refusal (stop_reason "refusal" carrying stop_details) re-enters the fallback chain only
+        when a content-policy fallback is configured; a plain refusal without stop_details, or
+        any response with nothing configured, is returned to the client unchanged.
+        """
+        from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
+            get_safeguard_refusal_stop_details,
         )
-        return False
+
+        if getattr(original_generic_function, "__name__", "") != "anthropic_messages":
+            return False
+        if get_safeguard_refusal_stop_details(response) is None:
+            return False
+        return self._has_content_policy_fallback(model, kwargs)
 
     def _get_healthy_deployments(self, model: str, parent_otel_span: Span | None):
         _all_deployments: list = []
@@ -12087,6 +12207,7 @@ class Router:
             if pre_routing_hook_response is not None:
                 model = pre_routing_hook_response.model
                 messages = pre_routing_hook_response.messages
+                record_pre_routing_selection(request_kwargs, model)
                 if pre_routing_hook_response.litellm_params:
                     accepted_tier_params: Final = self._tier_params_the_target_accepts(
                         model, pre_routing_hook_response.litellm_params, request_kwargs
@@ -12202,6 +12323,7 @@ class Router:
             if pre_routing_hook_response is not None:
                 model = pre_routing_hook_response.model
                 messages = pre_routing_hook_response.messages
+                record_pre_routing_selection(request_kwargs, model)
                 if pre_routing_hook_response.litellm_params:
                     accepted_tier_params: Final = self._tier_params_the_target_accepts(
                         model, pre_routing_hook_response.litellm_params, request_kwargs
diff --git a/litellm/router_utils/fallback_event_handlers.py b/litellm/router_utils/fallback_event_handlers.py
index d2842294a08..3d37ca216a7 100644
--- a/litellm/router_utils/fallback_event_handlers.py
+++ b/litellm/router_utils/fallback_event_handlers.py
@@ -214,6 +214,91 @@ def _check_stripped_model_group(model_group: str, fallback_key: str) -> bool:
     return False
 
 
+PRE_ROUTING_SELECTED_MODEL_KEY: Final = "pre_routing_selected_model"
+_ROUTER_METADATA_BUCKETS: Final = ("metadata", "litellm_metadata")
+
+
+def record_pre_routing_selection(request_kwargs: Mapping[str, Any] | None, selected_model: str) -> None:
+    """
+    Remember which model a pre-routing hook picked, so fallback lookup can key off it.
+
+    Fallback resolution runs on an outer kwargs dict that ``**kwargs`` already copied, so
+    writing the model there is invisible by the time routing picks a tier. The metadata
+    buckets are nested dicts shared by reference across those copies, which is how the
+    router already carries values back up.
+
+    The write goes through the proxy-internal bucket resolver, never into both buckets:
+    on /v1/messages the top-level ``metadata`` dict is the provider's own request field,
+    so a blanket write would forward the tier stamp upstream.
+    """
+    from litellm.litellm_core_utils.core_helpers import get_metadata_variable_name_from_kwargs
+
+    if request_kwargs is None:
+        return
+    bucket: Final = request_kwargs.get(get_metadata_variable_name_from_kwargs(request_kwargs))
+    if isinstance(bucket, dict):
+        bucket[PRE_ROUTING_SELECTED_MODEL_KEY] = selected_model
+
+
+def clear_pre_routing_selection(request_kwargs: Mapping[str, object] | None) -> None:
+    """
+    Drop any selection the router did not make itself on this hop.
+
+    The buckets carry whatever the caller sent, so an inbound value is the caller
+    choosing a fallback chain rather than the router choosing a tier. A fallback hop
+    also inherits the previous hop's selection, which would key its own failure off
+    the tier that already failed. Clearing at the start of every hop leaves only a
+    value the pre-routing hook wrote while routing that hop.
+    """
+    if request_kwargs is None:
+        return
+    for bucket in (request_kwargs.get(name) for name in _ROUTER_METADATA_BUCKETS):
+        if isinstance(bucket, dict) and PRE_ROUTING_SELECTED_MODEL_KEY in bucket:
+            del bucket[PRE_ROUTING_SELECTED_MODEL_KEY]
+
+
+def get_pre_routing_selection(kwargs: Mapping[str, Any]) -> str | None:
+    """The model a pre-routing hook selected for this request, if one did."""
+    buckets: Final = (kwargs.get(name) for name in _ROUTER_METADATA_BUCKETS)
+    selections: Final = (bucket.get(PRE_ROUTING_SELECTED_MODEL_KEY) for bucket in buckets if isinstance(bucket, dict))
+    return next((selected for selected in selections if isinstance(selected, str) and selected), None)
+
+
+def fallback_lookup_groups(kwargs: Mapping[str, Any], model_group: str | None) -> tuple[str, ...]:
+    """
+    Ordered keys for resolving a fallback chain: the tier a pre-routing hook selected wins,
+    and the requested group still resolves when no tier-keyed chain exists, so configs keyed
+    on the router name (the documented contract) keep working behind auto-routers.
+    """
+    ordered: Final = (get_pre_routing_selection(kwargs), model_group)
+    return tuple(dict.fromkeys(group for group in ordered if group))
+
+
+def _resolved_a_specific_chain(
+    fallbacks: list[Any],  # mutable-ok: mirrors get_fallback_model_group's contract
+    result: tuple[list[str] | None, int | None],  # mutable-ok: mirrors get_fallback_model_group's contract
+) -> bool:
+    resolved, generic_idx = result
+    if resolved is None:
+        return False
+    return generic_idx is None or resolved is not fallbacks[generic_idx]["*"]
+
+
+def get_fallback_model_group_for_lookup_groups(
+    fallbacks: list[Any],  # mutable-ok: mirrors get_fallback_model_group's contract
+    lookup_groups: tuple[str, ...],
+) -> tuple[list[str] | None, int | None]:  # mutable-ok: mirrors get_fallback_model_group's contract
+    """
+    First lookup group with a specifically-keyed chain wins; the generic "*" chain applies
+    only after every group missed, so a catch-all cannot shadow a later group's own chain.
+    """
+    results: Final = tuple(get_fallback_model_group(fallbacks=fallbacks, model_group=group) for group in lookup_groups)
+    specific: Final = next((result for result in results if _resolved_a_specific_chain(fallbacks, result)), None)
+    if specific is not None:
+        return specific
+    return next((result for result in results if result[0] is not None), (None, None))
+
+
 def get_fallback_model_group(fallbacks: list[Any], model_group: str) -> tuple[list[str] | None, int | None]:
     """
     Returns:
diff --git a/litellm/types/llms/anthropic_messages/anthropic_response.py b/litellm/types/llms/anthropic_messages/anthropic_response.py
index 42ca3fd6d4b..4fe1dafc73b 100644
--- a/litellm/types/llms/anthropic_messages/anthropic_response.py
+++ b/litellm/types/llms/anthropic_messages/anthropic_response.py
@@ -78,6 +78,16 @@ class AnthropicUsage(TypedDict, total=False):
     server_tool_use: NotRequired[ReadOnly[ServerToolUsage]]
 
 
+class AnthropicStopDetails(TypedDict, total=False):
+    """
+    Safeguard verdict accompanying a `stop_reason: "refusal"` response:
+    https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback
+    """
+
+    category: ReadOnly[str | None]
+    explanation: ReadOnly[str | None]
+
+
 class AnthropicMessagesResponse(TypedDict, total=False):
     """
     Anthropic Messages API Response: https://docs.anthropic.com/en/api/messages
@@ -90,7 +100,8 @@ class AnthropicMessagesResponse(TypedDict, total=False):
     id: str
     model: str | None  # This represents the Model type from Anthropic
     role: Literal["assistant"] | None
-    stop_reason: Literal["end_turn", "max_tokens", "stop_sequence", "tool_use"] | None
+    stop_reason: Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal"] | None
+    stop_details: NotRequired[ReadOnly[AnthropicStopDetails | None]]
     stop_sequence: str | None
     type: Literal["message"] | None
     usage: AnthropicUsage | None
diff --git a/tests/router_unit_tests/test_router_anthropic_messages_fallback.py b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py
new file mode 100644
index 00000000000..0c4d1dfc21e
--- /dev/null
+++ b/tests/router_unit_tests/test_router_anthropic_messages_fallback.py
@@ -0,0 +1,402 @@
+"""
+Unit tests for safeguard-refusal fallback on the /v1/messages router surface.
+
+An Anthropic safeguard refusal is an HTTP 200 whose body carries
+stop_reason "refusal" plus a stop_details object; the router converts it
+into a ContentPolicyViolationError so the content-policy fallback chain
+runs, but only when a matching fallback is configured. A plain refusal
+without stop_details, or any refusal with nothing configured, must reach
+the client byte-identical.
+
+The upstream is faked at the HTTP boundary by intercepting the third-party
+transport (httpx.AsyncClient.send), so requests run litellm's real
+transformation, allowlist, and streaming pipeline end to end.
+"""
+
+import json
+from typing import Any, AsyncIterator
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from litellm import Router
+from litellm.router_utils.fallback_event_handlers import (
+    PRE_ROUTING_SELECTED_MODEL_KEY,
+    record_pre_routing_selection,
+)
+
+REFUSAL_RESPONSE: dict[str, Any] = {
+    "id": "msg_refusal",
+    "type": "message",
+    "role": "assistant",
+    "model": "claude-fable-5",
+    "content": [],
+    "stop_reason": "refusal",
+    "stop_sequence": None,
+    "stop_details": {"category": "cyber", "explanation": "flagged"},
+    "usage": {"input_tokens": 25, "output_tokens": 1},
+}
+
+PLAIN_REFUSAL_RESPONSE: dict[str, Any] = {k: v for k, v in REFUSAL_RESPONSE.items() if k != "stop_details"}
+
+OK_RESPONSE: dict[str, Any] = {
+    "id": "msg_ok",
+    "type": "message",
+    "role": "assistant",
+    "model": "claude-opus-5",
+    "content": [{"type": "text", "text": "hello"}],
+    "stop_reason": "end_turn",
+    "stop_sequence": None,
+    "usage": {"input_tokens": 25, "output_tokens": 2},
+}
+
+
+def _sse(event: str, data: dict[str, Any]) -> bytes:
+    return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode()
+
+
+REFUSAL_STREAM_FRAMES: tuple[bytes, ...] = (
+    _sse("message_start", {"type": "message_start", "message": {**REFUSAL_RESPONSE, "stop_reason": None}}),
+    _sse(
+        "message_delta",
+        {
+            "type": "message_delta",
+            "delta": {"stop_reason": "refusal", "stop_details": {"category": "cyber"}},
+            "usage": {"output_tokens": 1},
+        },
+    ),
+    _sse("message_stop", {"type": "message_stop"}),
+)
+
+OK_STREAM_FRAMES: tuple[bytes, ...] = (
+    _sse("message_start", {"type": "message_start", "message": {**OK_RESPONSE, "stop_reason": None}}),
+    _sse(
+        "content_block_delta",
+        {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hello"}},
+    ),
+    _sse("message_stop", {"type": "message_stop"}),
+)
+
+
+def _split_frames_mid_data_line(frames: tuple[bytes, ...]) -> tuple[bytes, ...]:
+    """Split each frame's data line in half, modeling a transport chunk boundary."""
+    return tuple(part for frame in frames for part in (frame[: len(frame) // 2], frame[len(frame) // 2 :]))
+
+
+class _FrameStream(httpx.AsyncByteStream):
+    def __init__(self, frames: tuple[bytes, ...]) -> None:
+        self._frames = frames
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        for frame in self._frames:
+            yield frame
+
+    async def aclose(self) -> None:
+        return None
+
+
+class FakeAnthropicUpstream:
+    """Intercepts the third-party transport (httpx.AsyncClient.send): refuses on fable
+    models, answers on others. The router deliberately does not forward caller-injected
+    clients, so the transport is the seam that exercises the real litellm pipeline."""
+
+    def __init__(
+        self,
+        refusal_body: dict[str, Any] = REFUSAL_RESPONSE,
+        refusal_frames: tuple[bytes, ...] = REFUSAL_STREAM_FRAMES,
+    ) -> None:
+        self.refusal_body = refusal_body
+        self.refusal_frames = refusal_frames
+        self.calls: list[str] = []
+        self.bodies: list[dict[str, Any]] = []
+
+    async def send(self, request: httpx.Request, **kwargs: Any) -> httpx.Response:
+        body = json.loads(request.content or b"{}")
+        model = body.get("model", "")
+        self.calls.append(model)
+        self.bodies.append(body)
+        refuses = "fable" in model
+        if body.get("stream"):
+            frames = self.refusal_frames if refuses else OK_STREAM_FRAMES
+            return httpx.Response(
+                200,
+                stream=_FrameStream(frames),
+                headers={"content-type": "text/event-stream"},
+                request=request,
+            )
+        return httpx.Response(200, json=self.refusal_body if refuses else OK_RESPONSE, request=request)
+
+    def install(self):
+        async def _send(_client: httpx.AsyncClient, request: httpx.Request, **kwargs: Any) -> httpx.Response:
+            return await self.send(request, **kwargs)
+
+        return patch("httpx.AsyncClient.send", new=_send)
+
+
+FABLE_TIER = {
+    "model_name": "fable-tier",
+    "litellm_params": {"model": "anthropic/claude-fable-5", "api_key": "sk-test"},
+}
+OPUS_TARGET = {
+    "model_name": "opus-target",
+    "litellm_params": {"model": "anthropic/claude-opus-5", "api_key": "sk-test"},
+}
+
+
+def _router(content_policy_fallbacks: list | None) -> Router:
+    return Router(model_list=[FABLE_TIER, OPUS_TARGET], content_policy_fallbacks=content_policy_fallbacks)
+
+
+async def _collect(stream: AsyncIterator[bytes]) -> bytes:
+    return b"".join([chunk async for chunk in stream])
+
+
+@pytest.mark.asyncio
+async def test_non_streaming_refusal_with_fallback_row_returns_fallback_response():
+    fake = FakeAnthropicUpstream()
+    router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
+
+    with fake.install():
+        response = await router.aanthropic_messages(
+            model="fable-tier", max_tokens=16, messages=[{"role": "user", "content": "hi"}]
+        )
+
+    assert response["stop_reason"] == "end_turn"
+    assert response["id"] == "msg_ok"
+    assert len(fake.calls) == 2
+    assert "claude-opus-5" in fake.calls[1]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "content_policy_fallbacks, upstream_body",
+    [
+        (None, REFUSAL_RESPONSE),
+        ([{"unrelated-group": ["opus-target"]}], REFUSAL_RESPONSE),
+        ([{"fable-tier": ["opus-target"]}], PLAIN_REFUSAL_RESPONSE),
+    ],
+    ids=["nothing-configured", "row-for-other-group", "refusal-without-stop-details"],
+)
+async def test_non_streaming_refusal_passes_through_untouched(content_policy_fallbacks, upstream_body):
+    fake = FakeAnthropicUpstream(refusal_body=upstream_body)
+    router = _router(content_policy_fallbacks=content_policy_fallbacks)
+
+    with fake.install():
+        response = await router.aanthropic_messages(
+            model="fable-tier", max_tokens=16, messages=[{"role": "user", "content": "hi"}]
+        )
+
+    assert response["stop_reason"] == "refusal"
+    assert response.get("stop_details") == upstream_body.get("stop_details")
+    assert len(fake.calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_streaming_refusal_with_fallback_row_streams_fallback_frames():
+    fake = FakeAnthropicUpstream()
+    router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
+
+    with fake.install():
+        stream = await router.aanthropic_messages(
+            model="fable-tier", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
+        )
+        body = await _collect(stream)
+
+    assert b'"refusal"' not in body
+    assert b"text_delta" in body
+    assert len(fake.calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_streaming_refusal_split_across_chunks_still_falls_back():
+    fake = FakeAnthropicUpstream(refusal_frames=_split_frames_mid_data_line(REFUSAL_STREAM_FRAMES))
+    router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
+
+    with fake.install():
+        stream = await router.aanthropic_messages(
+            model="fable-tier", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
+        )
+        body = await _collect(stream)
+
+    assert b'"refusal"' not in body
+    assert b"text_delta" in body
+    assert len(fake.calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_streaming_refusal_without_fallback_row_passes_frames_through():
+    fake = FakeAnthropicUpstream()
+    router = _router(content_policy_fallbacks=None)
+
+    with fake.install():
+        stream = await router.aanthropic_messages(
+            model="fable-tier", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
+        )
+        body = await _collect(stream)
+
+    assert b'"stop_reason": "refusal"' in body
+    assert b"stop_details" in body
+    assert len(fake.calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_streaming_refusal_on_routed_tier_matches_tier_keyed_row_without_inbound_metadata():
+    """The pre-routing hook's tier stamp must reach the mid-stream fallback lookup even when the
+    request carries no metadata bucket at all (the snapshot is taken before the request runs)."""
+    fake = FakeAnthropicUpstream()
+    smart_router = {
+        "model_name": "smart-router",
+        "litellm_params": {
+            "model": "auto_router/complexity_router",
+            "complexity_router_config": {
+                "tiers": {"SIMPLE": "fable-tier", "MEDIUM": "fable-tier", "COMPLEX": "fable-tier"}
+            },
+            "complexity_router_default_model": "fable-tier",
+        },
+        "model_info": {"id": "router-1", "db_model": True},
+    }
+    router = Router(
+        model_list=[FABLE_TIER, OPUS_TARGET, smart_router],
+        content_policy_fallbacks=[{"fable-tier": ["opus-target"]}],
+        ignore_invalid_deployments=True,
+    )
+
+    with fake.install():
+        stream = await router.aanthropic_messages(
+            model="smart-router", max_tokens=16, stream=True, messages=[{"role": "user", "content": "hi"}]
+        )
+        body = await _collect(stream)
+
+    assert b'"refusal"' not in body
+    assert b"text_delta" in body
+    assert len(fake.calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_caller_forged_tier_stamp_cannot_pick_the_streaming_fallback_chain():
+    fake = FakeAnthropicUpstream()
+    router = _router(content_policy_fallbacks=[{"forged-tier": ["opus-target"]}])
+
+    with fake.install():
+        stream = await router.aanthropic_messages(
+            model="fable-tier",
+            max_tokens=16,
+            stream=True,
+            messages=[{"role": "user", "content": "hi"}],
+            litellm_metadata={PRE_ROUTING_SELECTED_MODEL_KEY: "forged-tier"},
+        )
+        body = await _collect(stream)
+
+    assert b'"stop_reason": "refusal"' in body
+    assert len(fake.calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_tier_stamp_never_reaches_provider_bound_metadata():
+    """On /v1/messages the top-level metadata dict is Anthropic's own request field, so the
+    routed-tier stamp must never appear in any upstream body even when the client sends one."""
+    fake = FakeAnthropicUpstream()
+    smart_router = {
+        "model_name": "smart-router",
+        "litellm_params": {
+            "model": "auto_router/complexity_router",
+            "complexity_router_config": {
+                "tiers": {"SIMPLE": "fable-tier", "MEDIUM": "fable-tier", "COMPLEX": "fable-tier"}
+            },
+            "complexity_router_default_model": "fable-tier",
+        },
+        "model_info": {"id": "router-1", "db_model": True},
+    }
+    router = Router(
+        model_list=[FABLE_TIER, OPUS_TARGET, smart_router],
+        content_policy_fallbacks=[{"fable-tier": ["opus-target"]}],
+        ignore_invalid_deployments=True,
+    )
+
+    with fake.install():
+        response = await router.aanthropic_messages(
+            model="smart-router",
+            max_tokens=16,
+            messages=[{"role": "user", "content": "hi"}],
+            metadata={"user_id": "u1"},
+        )
+
+    assert response["stop_reason"] == "end_turn"
+    assert len(fake.bodies) == 2
+    for body in fake.bodies:
+        assert body.get("metadata") == {"user_id": "u1"}
+
+
+def test_record_pre_routing_selection_writes_only_the_internal_bucket():
+    """The Anthropic request's own metadata field must never carry the tier stamp."""
+    kwargs = {"metadata": {"user_id": "u1"}, "litellm_metadata": {}}
+
+    record_pre_routing_selection(kwargs, "tier-x")
+
+    assert kwargs["litellm_metadata"] == {PRE_ROUTING_SELECTED_MODEL_KEY: "tier-x"}
+    assert kwargs["metadata"] == {"user_id": "u1"}
+
+
+def test_refusal_gate_keys_on_pre_routing_tier_stamp():
+    router = _router(content_policy_fallbacks=[{"tier-group": ["opus-target"]}])
+
+    def anthropic_messages(**kwargs: Any) -> None:
+        return None
+
+    refusal_kwargs = {"litellm_metadata": {PRE_ROUTING_SELECTED_MODEL_KEY: "tier-group"}}
+    assert (
+        router._should_raise_anthropic_refusal_error(
+            model="router-group",
+            original_generic_function=anthropic_messages,
+            response=dict(REFUSAL_RESPONSE),
+            kwargs=refusal_kwargs,
+        )
+        is True
+    )
+    assert (
+        router._should_raise_anthropic_refusal_error(
+            model="router-group",
+            original_generic_function=anthropic_messages,
+            response=dict(REFUSAL_RESPONSE),
+            kwargs={},
+        )
+        is False
+    )
+
+
+def test_has_content_policy_fallback_default_fallbacks_arm():
+    router = Router(model_list=[OPUS_TARGET], fallbacks=[{"*": ["opus-target"]}])
+
+    assert router._has_content_policy_fallback("any-group", {}) is True
+    assert router._has_content_policy_fallback("any-group", {"content_policy_fallbacks": [{"other": ["x"]}]}) is False
+
+
+def test_get_fallback_model_group_for_lookup_groups_orders_tier_before_requested():
+    router = _router(content_policy_fallbacks=None)
+    fallbacks = [{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}]
+
+    assert router._get_fallback_model_group_for_lookup_groups(
+        fallbacks=fallbacks, lookup_groups=("tier1", "smart-router")
+    ) == ["backup-a"]
+    assert router._get_fallback_model_group_for_lookup_groups(
+        fallbacks=fallbacks, lookup_groups=("tier9", "smart-router")
+    ) == ["backup-b"]
+    assert router._get_fallback_model_group_for_lookup_groups(fallbacks=fallbacks, lookup_groups=()) is None
+
+
+def test_refusal_gate_ignores_other_generic_call_types():
+    router = _router(content_policy_fallbacks=[{"fable-tier": ["opus-target"]}])
+
+    def aresponses(**kwargs: Any) -> None:
+        return None
+
+    assert (
+        router._should_raise_anthropic_refusal_error(
+            model="fable-tier",
+            original_generic_function=aresponses,
+            response=dict(REFUSAL_RESPONSE),
+            kwargs={},
+        )
+        is False
+    )
diff --git a/tests/test_litellm/router_utils/test_fallback_event_handlers.py b/tests/test_litellm/router_utils/test_fallback_event_handlers.py
index 8336926c050..894b2d9e74f 100644
--- a/tests/test_litellm/router_utils/test_fallback_event_handlers.py
+++ b/tests/test_litellm/router_utils/test_fallback_event_handlers.py
@@ -11,7 +11,10 @@ from litellm.router_utils.fallback_event_handlers import (
     AttemptedFallbackTargets,
     _trigger_cooldown_for_failed_deployment,
     fallback_attempt_key,
+    clear_pre_routing_selection,
     get_fallback_model_group,
+    get_pre_routing_selection,
+    record_pre_routing_selection,
     run_async_fallback,
 )
 
@@ -1090,3 +1093,119 @@ async def test_run_async_fallback_preserves_original_model_group_on_nested_fallb
     metadata = router.received_kwargs["metadata"]
     assert metadata["attempted_fallbacks"] == 2
     assert metadata["original_model_group"] == "primary-model"
+
+
+class TestPreRoutingSelectionCarriesToFallbacks:
+    """#38832: a complexity/auto router picks a tier behind the router name, but fallback
+    lookup kept using the router name, so the tier's configured chain never ran."""
+
+    def test_selection_is_recorded_in_the_metadata_bucket(self):
+        kwargs = {"model": "smart-router", "metadata": {}}
+        record_pre_routing_selection(kwargs, "tier1")
+        assert kwargs["metadata"]["pre_routing_selected_model"] == "tier1"
+        assert get_pre_routing_selection(kwargs) == "tier1"
+
+    def test_selection_is_recorded_in_the_litellm_metadata_bucket(self):
+        kwargs = {"model": "smart-router", "litellm_metadata": {}}
+        record_pre_routing_selection(kwargs, "tier2")
+        assert get_pre_routing_selection(kwargs) == "tier2"
+
+    def test_a_bucket_survives_the_kwargs_copy_that_fallbacks_run_on(self):
+        """The bucket is shared by reference, which is the whole reason this works."""
+        outer = {"model": "smart-router", "metadata": {}}
+        inner = {**outer}
+        record_pre_routing_selection(inner, "tier1")
+        assert get_pre_routing_selection(outer) == "tier1"
+
+    def test_no_selection_reads_as_none(self):
+        assert get_pre_routing_selection({"model": "smart-router", "metadata": {}}) is None
+        assert get_pre_routing_selection({"model": "smart-router"}) is None
+
+    def test_missing_kwargs_is_a_no_op(self):
+        """A caller with no kwargs must not raise, and must not leak the selection anywhere."""
+        record_pre_routing_selection(None, "tier1")
+
+        assert get_pre_routing_selection({}) is None
+
+    def test_a_non_dict_bucket_is_ignored(self):
+        kwargs = {"model": "smart-router", "metadata": "not-a-dict"}
+        record_pre_routing_selection(kwargs, "tier1")
+        assert get_pre_routing_selection(kwargs) is None
+
+    def test_fallbacks_resolve_against_the_selected_tier(self):
+        """The lookup the router performs, keyed on the tier rather than the router name."""
+        fallbacks = [{"tier1": ["backup-a", "backup-b"]}, {"tier2": ["backup-c"]}]
+        assert get_fallback_model_group(fallbacks=fallbacks, model_group="tier1")[0] == ["backup-a", "backup-b"]
+        assert get_fallback_model_group(fallbacks=fallbacks, model_group="smart-router")[0] is None
+
+
+class TestPreRoutingSelectionIsPerHop:
+    """#38832 review: the buckets also carry whatever the caller sent, and a fallback hop
+    inherits the previous hop's tier, so a hop must start without a selection."""
+
+    def test_a_caller_supplied_selection_is_dropped(self):
+        kwargs = {"model": "plain", "metadata": {"pre_routing_selected_model": "tier1"}}
+
+        clear_pre_routing_selection(kwargs)
+
+        assert get_pre_routing_selection(kwargs) is None
+        assert "pre_routing_selected_model" not in kwargs["metadata"]
+
+    def test_both_buckets_are_cleared(self):
+        kwargs = {
+            "metadata": {"pre_routing_selected_model": "tier1"},
+            "litellm_metadata": {"pre_routing_selected_model": "tier2"},
+        }
+
+        clear_pre_routing_selection(kwargs)
+
+        assert get_pre_routing_selection(kwargs) is None
+
+    def test_the_rest_of_the_bucket_is_left_alone(self):
+        kwargs = {"metadata": {"pre_routing_selected_model": "tier1", "tags": ["a"]}}
+
+        clear_pre_routing_selection(kwargs)
+
+        assert kwargs["metadata"] == {"tags": ["a"]}
+
+    def test_clearing_is_a_no_op_without_a_usable_bucket(self):
+        kwargs = {"model": "plain", "metadata": "not-a-dict"}
+
+        clear_pre_routing_selection(None)
+        clear_pre_routing_selection(kwargs)
+
+        assert kwargs == {"model": "plain", "metadata": "not-a-dict"}
+
+    def test_a_selection_recorded_after_clearing_is_kept(self):
+        """Clearing runs before routing, so the hook's own write must survive it."""
+        kwargs = {"model": "smart-router", "metadata": {"pre_routing_selected_model": "stale"}}
+
+        clear_pre_routing_selection(kwargs)
+        record_pre_routing_selection(kwargs, "tier1")
+
+        assert get_pre_routing_selection(kwargs) == "tier1"
+
+
+class TestOrderedFallbackLookupGroups:
+    def test_tier_first_then_requested_group_deduped(self):
+        from litellm.router_utils.fallback_event_handlers import (
+            PRE_ROUTING_SELECTED_MODEL_KEY,
+            fallback_lookup_groups,
+        )
+
+        kwargs = {"litellm_metadata": {PRE_ROUTING_SELECTED_MODEL_KEY: "tier1"}}
+        assert fallback_lookup_groups(kwargs, "smart-router") == ("tier1", "smart-router")
+        assert fallback_lookup_groups(kwargs, "tier1") == ("tier1",)
+        assert fallback_lookup_groups({}, "smart-router") == ("smart-router",)
+        assert fallback_lookup_groups({}, None) == ()
+
+    def test_first_resolving_group_wins_and_generic_idx_survives_a_miss(self):
+        from litellm.router_utils.fallback_event_handlers import (
+            get_fallback_model_group_for_lookup_groups,
+        )
+
+        fallbacks = [{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}, {"*": ["backup-c"]}]
+        assert get_fallback_model_group_for_lookup_groups(fallbacks, ("tier1", "smart-router")) == (["backup-a"], None)
+        assert get_fallback_model_group_for_lookup_groups(fallbacks, ("tier9", "smart-router")) == (["backup-b"], None)
+        assert get_fallback_model_group_for_lookup_groups(fallbacks, ("tier9", "no-such")) == (["backup-c"], 2)
+        assert get_fallback_model_group_for_lookup_groups([{"tier1": ["backup-a"]}], ("no", "nope")) == (None, None)
diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py
index 44c1cdbff06..84f6344be35 100644
--- a/tests/test_litellm/test_router.py
+++ b/tests/test_litellm/test_router.py
@@ -11595,3 +11595,138 @@ class TestTierParamsTheTargetAccepts:
         accepted = router._tier_params_the_target_accepts("no-such-group", {"reasoning_effort": "max"}, {})
 
         assert accepted == {"reasoning_effort": "max"}
+
+
+class TestPreRoutingTierDrivesFallbacks:
+    """#38832: a complexity/auto router picks a tier behind the router name, but fallback
+    lookup stayed on the router name, so the tier's configured chain never ran and a
+    provider failure on the tier's first hop was returned to the client."""
+
+    class _TierRouter(litellm.Router):
+        async def async_pre_routing_hook(
+            self, model, request_kwargs, messages=None, input=None, specific_deployment=False
+        ):
+            from litellm.types.router import PreRoutingHookResponse
+
+            if model == "smart-router":
+                return PreRoutingHookResponse(model="tier1", messages=messages)
+            return None
+
+    @classmethod
+    def _router(cls, fallbacks) -> "litellm.Router":
+        return cls._TierRouter(
+            model_list=[
+                {
+                    "model_name": "smart-router",
+                    "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-x"},
+                },
+                {
+                    "model_name": "tier1",
+                    "litellm_params": {
+                        "model": "openai/gpt-4o-mini",
+                        "api_key": "sk-x",
+                        "mock_response": "litellm.RateLimitError",
+                    },
+                },
+                {
+                    "model_name": "backup-a",
+                    "litellm_params": {
+                        "model": "openai/gpt-4o-mini",
+                        "api_key": "sk-x",
+                        "mock_response": "from backup-a",
+                    },
+                },
+                {
+                    "model_name": "backup-b",
+                    "litellm_params": {
+                        "model": "openai/gpt-4o-mini",
+                        "api_key": "sk-x",
+                        "mock_response": "from backup-b",
+                    },
+                },
+                {
+                    "model_name": "failing-backup",
+                    "litellm_params": {
+                        "model": "openai/gpt-4o-mini",
+                        "api_key": "sk-x",
+                        "mock_response": "litellm.RateLimitError",
+                    },
+                },
+                {
+                    "model_name": "plain",
+                    "litellm_params": {
+                        "model": "openai/gpt-4o-mini",
+                        "api_key": "sk-x",
+                        "mock_response": "litellm.RateLimitError",
+                    },
+                },
+            ],
+            fallbacks=fallbacks,
+            num_retries=0,
+        )
+
+    @pytest.mark.asyncio
+    async def test_the_selected_tier_fallback_chain_runs(self):
+        router = self._router([{"tier1": ["backup-a"]}])
+
+        response = await router.acompletion(
+            model="smart-router", messages=[{"role": "user", "content": "hi"}]
+        )
+
+        assert response.choices[0].message.content == "from backup-a"
+
+    @pytest.mark.asyncio
+    async def test_a_chain_keyed_on_the_router_name_is_not_used(self):
+        """The router name has no chain of its own, so nothing should rescue this call."""
+        router = self._router([{"tier2": ["backup-a"]}])
+
+        with pytest.raises(litellm.RateLimitError):
+            await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
+
+    @pytest.mark.asyncio
+    async def test_a_chain_keyed_on_the_router_name_rescues_when_no_tier_chain_exists(self):
+        """The documented contract: configs keyed on the requested name keep working behind auto-routers."""
+        router = self._router([{"smart-router": ["backup-a"]}])
+
+        response = await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
+
+        assert response.choices[0].message.content == "from backup-a"
+
+    @pytest.mark.asyncio
+    async def test_the_tier_chain_wins_over_the_router_name_chain(self):
+        router = self._router([{"tier1": ["backup-a"]}, {"smart-router": ["backup-b"]}])
+
+        response = await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
+
+        assert response.choices[0].message.content == "from backup-a"
+
+    @pytest.mark.asyncio
+    async def test_a_request_without_a_pre_routing_hook_still_uses_its_own_group(self):
+        router = self._router([{"tier1": ["backup-a"]}])
+
+        response = await router.acompletion(
+            model="tier1", messages=[{"role": "user", "content": "hi"}]
+        )
+
+        assert response.choices[0].message.content == "from backup-a"
+
+    @pytest.mark.asyncio
+    async def test_a_caller_cannot_pick_the_chain_by_sending_the_selection(self):
+        """The metadata bucket carries caller-supplied keys, so only the hook may set the tier."""
+        router = self._router([{"tier1": ["backup-a"]}])
+
+        with pytest.raises(litellm.RateLimitError):
+            await router.acompletion(
+                model="plain",
+                messages=[{"role": "user", "content": "hi"}],
+                metadata={"pre_routing_selected_model": "tier1"},
+            )
+
+    @pytest.mark.asyncio
+    async def test_each_fallback_hop_resolves_its_own_chain(self):
+        """The second hop must key off the group it is running, not the tier that failed."""
+        router = self._router([{"tier1": ["failing-backup"]}, {"failing-backup": ["backup-b"]}])
+
+        response = await router.acompletion(model="smart-router", messages=[{"role": "user", "content": "hi"}])
+
+        assert response.choices[0].message.content == "from backup-b"