fix(sdk): stop global model_fallbacks from recursing on every completion call (#51817)

* fix(sdk): stop global model_fallbacks from recursing on every completion call

With litellm.model_fallbacks set, acompletion and completion read the global
chain on every call, including the calls the fallback runner itself makes for
each attempt. Each attempt started a new fallback run, so the calls grew into a
tree until Python hit RecursionError and no provider was ever reached.

The fallback runner now marks every attempt with an internal _litellm_ kwarg,
and one resolver in fallback_utils decides the chain for a call. A marked
attempt gets no chain. Otherwise a per-call fallbacks value wins, including an
empty list to opt out, and the global list is used only when the call sets
none. An empty chain skips the runner. The resolved chain is what the runner
gets, so completion's sync path passes it explicitly. The marker is dropped
from provider params and cache keys like every other _litellm_ kwarg

* fix(fallbacks): keep each provider's credentials, typed errors, sync streams and one success log

A fallback to another provider drops the original request's credentials and
endpoint, and a fallback to another model drops per-call custom pricing.
When every attempt fails, the original model's typed error is re-raised.
The sync runner calls litellm.completion directly so sync streams work, and
the sync wrapper leaves the success log to the attempt inside the runner.
Health checks never fall back, and every attempt keeps the caller's
litellm_call_id.

* fix(fallbacks): set up fallbacks only when tried and keep health checks off the global list in every mode

Each fallback is now built only when the runner reaches it, using the provider
lookup that never starts a ChatGPT or GitHub Copilot device login, so an unused
fallback cannot block a healthy call

Health checks run inside a scope that turns the global model_fallbacks off,
which also covers modes bridged to chat. The per-site fallbacks=() overrides
are removed

The sync logging test now forces the inner and outer success handlers to
overlap with a barrier instead of sleeping

* fix(fallbacks): keep guardrails on sync attempts and keep the global list out of Router calls

The sync fallback runner now runs the deployment pre and post call hooks around every attempt, so guardrails a sync call names still run on each model, and it skips the post hook for streams like the async wrapper does

Router deployment calls run with the global list turned off, so on the proxy the Router owns fallbacks and a fallback answer is never priced or logged as the failed deployment

A fallback to another provider now also drops the request's headers and litellm_credential_name, and an original model whose provider cannot be looked up counts as a switch

When every model fails the caller gets main's documented "All fallback attempts failed" error again

* fix(fallbacks): keep custom provider settings for unprefixed fallbacks and never cache a rejected answer

An unprefixed fallback whose name resolves to no provider on its own now inherits the request's custom_llm_provider, so a fallback on the same OpenAI-compatible server keeps its api_base and key as it did on main

Sync fallback attempts no longer write to the cache, since their guardrail runs after the inner call returns. The outer call still caches the answer that passed, so a rejected answer never lands in the cache

* fix(fallbacks): run a sync attempt's output guardrail before it is cached or logged

The sync runner ran the post-call guardrail after litellm.completion returned, so the attempt's success log already held the raw answer and a rejected answer was logged as a success. The sync wrapper now runs the guardrail for a fallback attempt at the same point the async wrapper does, before the cache write and the success log, which also replaces the no-store workaround

* fix(fallbacks): keep search grants, deployment hooks and logs intact

Router deployment calls wrap their handler to turn the global list off, which hid asearch from the search fallback grant check. The check now unwraps the handler down to asearch

A failed sync fallback attempt runs the post-call failure deployment hook again, as main did through its async detour

atext_completion through the global list uses the async runner so it logs one success, and check_valid_key turns the global list off so a fallback answer never makes a bad key look valid

* fix(proxy): keep the global fallback list off on direct SDK calls

Pass-through models and routes the Router does not serve reach litellm directly, so the global list ran there with no check that the caller's key may use the fallback model. Those calls now turn the global list off, as Router deployment calls already do

* fix(fallbacks): price a fallback by its own model and skip the proxy global list

A fallback to another model kept the original call's base_model, so its answer was billed at the original model's rates, as on main. base_model now drops with the other per-call pricing params

The proxy turns the global fallback list off on every model call, so litellm_settings.model_fallbacks now logs a startup warning that points to router_settings.fallbacks and is not applied

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

litellm/litellm_core_utils/clientside_credential_handler.py

@@ -82,6 +82,7 @@ def _admin_config_fields_to_clear_on_base_override() -> list[str]:
 
 
 _ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE: Final = _admin_config_fields_to_clear_on_base_override()
+UPSTREAM_BOUND_PARAMS: Final = frozenset((*clientside_credential_keys, *_ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE))
 
 
 def is_clientside_credential(request_kwargs: dict) -> bool:

litellm/litellm_core_utils/fallback_utils.py

@@ -1,85 +1,241 @@
-from typing import Final
+import functools
+import inspect
+from collections.abc import Awaitable, Callable, Generator, Iterator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Final, ParamSpec, TypeAlias, TypeVar
+
+from pydantic import TypeAdapter
 
 import litellm
 from litellm._logging import verbose_logger
 from litellm._uuid import uuid
+from litellm.constants import INTERNAL_KWARG_PREFIX
 from litellm.litellm_core_utils.add_retry_fallback_headers import (
     add_fallback_headers_to_response,
 )
+from litellm.litellm_core_utils.asyncify import run_async_function
+from litellm.litellm_core_utils.clientside_credential_handler import UPSTREAM_BOUND_PARAMS
 from litellm.litellm_core_utils.core_helpers import (
     filter_internal_params,
     safe_deep_copy,
 )
+from litellm.types.utils import CallTypes, CustomPricingLiteLLMParams
 
-from .asyncify import run_async_function
+FALLBACK_ATTEMPT_KWARG: Final = f"{INTERNAL_KWARG_PREFIX}fallback_attempt"
+
+FallbackChain: TypeAlias = tuple[str | dict[str, object], ...]
+_FALLBACK_CHAIN: Final = TypeAdapter(FallbackChain)
+_REQUEST_KWARGS: Final = TypeAdapter(dict[str, object])
+_NO_OVERRIDES: Final[Mapping[str, object]] = MappingProxyType({})
+_RUNNER_PARAMS: Final = frozenset({"kwargs", "fallbacks", "acompletion", "model", "litellm_logging_obj"})
+_PROVIDER_BOUND_PARAMS: Final = UPSTREAM_BOUND_PARAMS | {
+    "custom_llm_provider",
+    "client",
+    "headers",
+    "litellm_credential_name",
+}
+_MODEL_BOUND_PARAMS: Final = frozenset(CustomPricingLiteLLMParams.model_fields) | {"base_model"}
+_GLOBAL_FALLBACKS_OFF: Final[ContextVar[bool]] = ContextVar("litellm_global_model_fallbacks_off", default=False)
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
 
 
-async def async_completion_with_fallbacks(**kwargs):
-    """
-    Asynchronously attempts completion with fallback models if the primary model fails.
+@contextmanager
+def without_global_fallbacks() -> Generator[None]:
+    token: Final = _GLOBAL_FALLBACKS_OFF.set(True)
+    try:
+        yield
+    finally:
+        _GLOBAL_FALLBACKS_OFF.reset(token)
 
-    Args:
-        **kwargs: Keyword arguments for completion, including:
-            - model (str): Primary model to use
-            - fallbacks (List[Union[str, dict]]): List of fallback models/configs
-            - Other completion parameters
 
-    Returns:
-        ModelResponse: The completion response from the first successful model
+async def _awaited(pending: Awaitable[object]) -> object:
+    return await pending
 
-    Raises:
-        Exception: If all models fail and no response is generated
-    """
-    # Extract and prepare parameters
-    nested_kwargs: Final = kwargs.pop("kwargs", {})
-    original_model: Final = kwargs["model"]
-    model = original_model
-    fallbacks: Final = [original_model] + nested_kwargs.pop("fallbacks", [])
-    kwargs.pop("acompletion", None)  # Remove to prevent keyword conflicts
-    litellm_call_id: Final = str(uuid.uuid4())
-    base_kwargs: Final = {**kwargs, **nested_kwargs, "litellm_call_id": litellm_call_id}
 
-    # fields to remove
-    base_kwargs.pop("model", None)  # Remove model as it will be set per fallback
-    litellm_logging_obj: Final = base_kwargs.pop("litellm_logging_obj", None)
+def without_global_fallbacks_around(endpoint: Callable[_P, _R]) -> Callable[_P, _R]:
+    if not inspect.iscoroutinefunction(endpoint):
 
-    # Try each fallback model
-    most_recent_exception_str: str | None = None
-    for attempted_fallbacks, fallback in enumerate(fallbacks):
-        try:
-            completion_kwargs = safe_deep_copy(base_kwargs)
-            # Handle dictionary fallback configurations
-            if isinstance(fallback, dict):
-                fallback_config = safe_deep_copy(dict(fallback))
-                model = fallback_config.pop("model", original_model)
-                completion_kwargs.update(fallback_config)
-            else:
-                model = fallback
+        @functools.wraps(endpoint)
+        def call(*args: _P.args, **kwargs: _P.kwargs) -> _R:  # kwargs-ok: endpoint payload
+            with without_global_fallbacks():
+                return endpoint(*args, **kwargs)
 
-            # Filter out internal parameters that shouldn't be sent to provider APIs
-            completion_kwargs = filter_internal_params(completion_kwargs)
+        return call
 
-            response = await litellm.acompletion(
-                **completion_kwargs,
-                model=model,
-                litellm_logging_obj=litellm_logging_obj,
-            )
+    @functools.wraps(endpoint)
+    async def acall(*args: _P.args, **kwargs: _P.kwargs) -> object:  # kwargs-ok: endpoint payload
+        with without_global_fallbacks():
+            result: Final = endpoint(*args, **kwargs)
+            return await _awaited(result) if inspect.isawaitable(result) else result
 
-            if response is not None:
-                return add_fallback_headers_to_response(
-                    response=response,
-                    attempted_fallbacks=attempted_fallbacks,
-                )
+    return acall  # pyright: ignore[reportReturnType]  # a coroutine endpoint's _R is the coroutine acall returns
 
-        except Exception as e:
-            verbose_logger.exception("Fallback attempt failed for model %s: %s", model, e)
-            most_recent_exception_str = str(e)
-            continue
 
-    raise Exception(
-        f"{most_recent_exception_str}. All fallback attempts failed. Enable verbose logging with `litellm.set_verbose=True` for details."
+def _configured_fallbacks(request_kwargs: Mapping[str, object]) -> object:
+    if request_kwargs.get(FALLBACK_ATTEMPT_KWARG) is True:
+        return None
+    per_call: Final = request_kwargs.get("fallbacks")
+    if per_call is not None:
+        return per_call
+    return None if _GLOBAL_FALLBACKS_OFF.get() else litellm.model_fallbacks
+
+
+def uses_fallbacks(request_kwargs: Mapping[str, object]) -> bool:
+    return bool(_configured_fallbacks(request_kwargs))
+
+
+def resolve_fallbacks(request_kwargs: Mapping[str, object]) -> FallbackChain | None:
+    configured: Final = _configured_fallbacks(request_kwargs)
+    return _FALLBACK_CHAIN.validate_python(configured) if configured else None
+
+
+@dataclass(frozen=True, slots=True)
+class _FallbackAttempt:
+    model: str
+    kwargs: Mapping[str, object]
+
+
+@dataclass(frozen=True, slots=True)
+class _FallbackPlan:
+    original_model: str
+    base_kwargs: Mapping[str, object]
+    fallbacks: FallbackChain
+    logging_obj: object
+
+
+def _provider_of(model: str, custom_llm_provider: object) -> str | None:
+    from litellm.litellm_core_utils.get_llm_provider_logic import resolve_model_and_provider_for_metadata
+
+    try:
+        return resolve_model_and_provider_for_metadata(
+            model, custom_llm_provider if isinstance(custom_llm_provider, str) else None
+        )[1]
+    except litellm.exceptions.BadRequestError:
+        return None
+
+
+def _inherited_kwargs(
+    base_kwargs: Mapping[str, object], original_model: str, model: str, custom_llm_provider: object
+) -> Mapping[str, object]:
+    if model == original_model:
+        return base_kwargs
+    requested_provider: Final = base_kwargs.get("custom_llm_provider")
+    original_provider: Final = _provider_of(original_model, requested_provider)
+    provider: Final = _provider_of(model, custom_llm_provider) or _provider_of(model, requested_provider)
+    switches_provider: Final = original_provider is None or original_provider != provider
+    dropped: Final = (_MODEL_BOUND_PARAMS | _PROVIDER_BOUND_PARAMS) if switches_provider else _MODEL_BOUND_PARAMS
+    return {key: value for key, value in base_kwargs.items() if key not in dropped}
+
+
+def _fallback_attempt(plan: _FallbackPlan, fallback: str | Mapping[str, object]) -> _FallbackAttempt:
+    overrides: Final = _NO_OVERRIDES if isinstance(fallback, str) else fallback
+    model: Final = fallback if isinstance(fallback, str) else str(overrides.get("model", plan.original_model))
+    inherited: Final = _inherited_kwargs(
+        plan.base_kwargs, plan.original_model, model, overrides.get("custom_llm_provider")
+    )
+    request: Final = filter_internal_params(safe_deep_copy({**inherited, **overrides}))
+    return _FallbackAttempt(
+        model=model,
+        kwargs={**request, "model": model, "litellm_logging_obj": plan.logging_obj, FALLBACK_ATTEMPT_KWARG: True},
     )
 
 
-def completion_with_fallbacks(**kwargs):
-    return run_async_function(async_function=async_completion_with_fallbacks, **kwargs)
+def _fallback_plan(runner_kwargs: Mapping[str, object]) -> _FallbackPlan:
+    nested_kwargs: Final = _REQUEST_KWARGS.validate_python(runner_kwargs.get("kwargs") or {})
+    merged: Final = {**runner_kwargs, **nested_kwargs}
+    return _FallbackPlan(
+        original_model=str(runner_kwargs["model"]),
+        base_kwargs={
+            **{key: value for key, value in merged.items() if key not in _RUNNER_PARAMS},
+            "litellm_call_id": merged.get("litellm_call_id") or str(uuid.uuid4()),
+        },
+        fallbacks=_FALLBACK_CHAIN.validate_python(nested_kwargs.get("fallbacks") or ()),
+        logging_obj=merged.get("litellm_logging_obj"),
+    )
+
+
+def _fallback_attempts(plan: _FallbackPlan) -> Iterator[_FallbackAttempt]:
+    return (_fallback_attempt(plan, fallback) for fallback in plan.fallbacks)
+
+
+def _all_attempts_failed(last_error: Exception) -> Exception:
+    return Exception(
+        f"{last_error}. All fallback attempts failed. Enable verbose logging with `litellm.set_verbose=True` for details."
+    )
+
+
+async def async_completion_with_fallbacks(**kwargs: object) -> object:
+    plan: Final = _fallback_plan(kwargs)
+    primary: Final = _fallback_attempt(plan, plan.original_model)
+    try:
+        return add_fallback_headers_to_response(
+            response=await litellm.acompletion(**primary.kwargs), attempted_fallbacks=0
+        )
+    except Exception as primary_error:  # noqa: BLE001  # every failed attempt moves on to the next fallback
+        verbose_logger.exception("Fallback attempt failed for model %s: %s", primary.model, primary_error)
+        last_error: Exception = primary_error  # rebind-ok: the newest failure names the final error
+        for attempted_fallbacks, attempt in enumerate(_fallback_attempts(plan), start=1):
+            try:
+                return add_fallback_headers_to_response(
+                    response=await litellm.acompletion(**attempt.kwargs), attempted_fallbacks=attempted_fallbacks
+                )
+            except Exception as fallback_error:  # noqa: BLE001  # any failed attempt moves on to the next fallback
+                verbose_logger.exception("Fallback attempt failed for model %s: %s", attempt.model, fallback_error)
+                last_error = fallback_error
+        raise _all_attempts_failed(last_error) from last_error
+
+
+def _is_fallback_attempt(request_kwargs: Mapping[str, object], call_type: str) -> bool:
+    return call_type == CallTypes.completion.value and request_kwargs.get(FALLBACK_ATTEMPT_KWARG) is True
+
+
+def checked_fallback_attempt(request_kwargs: Mapping[str, object], call_type: str, response: object) -> object:
+    if not _is_fallback_attempt(request_kwargs, call_type):
+        return response
+    from litellm.utils import async_post_call_success_deployment_hook
+
+    return run_async_function(
+        async_post_call_success_deployment_hook, request_kwargs, response, CallTypes.completion, include_v2=True
+    )
+
+
+def report_failed_sync_fallback_attempt(request_kwargs: Mapping[str, object], call_type: str, error: Exception) -> None:
+    if not _is_fallback_attempt(request_kwargs, call_type) or request_kwargs.get("acompletion") is True:
+        return
+    from litellm.utils import async_post_call_failure_deployment_hook
+
+    run_async_function(async_post_call_failure_deployment_hook, request_kwargs, error, call_type, include_v2=True)
+
+
+def _completion_with_pre_call_hooks(request_kwargs: Mapping[str, object]) -> object:
+    from litellm.utils import async_pre_call_deployment_hook
+
+    request: Final = _REQUEST_KWARGS.validate_python(
+        run_async_function(async_pre_call_deployment_hook, request_kwargs, CallTypes.completion.value, include_v2=True)
+    )
+    return litellm.completion(**request)
+
+
+def completion_with_fallbacks(**kwargs: object) -> object:
+    plan: Final = _fallback_plan(kwargs)
+    primary: Final = _fallback_attempt(plan, plan.original_model)
+    try:
+        return add_fallback_headers_to_response(
+            response=_completion_with_pre_call_hooks(primary.kwargs), attempted_fallbacks=0
+        )
+    except Exception as primary_error:  # noqa: BLE001  # every failed attempt moves on to the next fallback
+        verbose_logger.exception("Fallback attempt failed for model %s: %s", primary.model, primary_error)
+        last_error: Exception = primary_error  # rebind-ok: the newest failure names the final error
+        for attempted_fallbacks, attempt in enumerate(_fallback_attempts(plan), start=1):
+            try:
+                return add_fallback_headers_to_response(
+                    response=_completion_with_pre_call_hooks(attempt.kwargs), attempted_fallbacks=attempted_fallbacks
+                )
+            except Exception as fallback_error:  # noqa: BLE001  # any failed attempt moves on to the next fallback
+                verbose_logger.exception("Fallback attempt failed for model %s: %s", attempt.model, fallback_error)
+                last_error = fallback_error
+        raise _all_attempts_failed(last_error) from last_error

litellm/main.py

@@ -243,6 +243,8 @@ from .litellm_core_utils.core_helpers import (
 from .litellm_core_utils.fallback_utils import (
     async_completion_with_fallbacks,
     completion_with_fallbacks,
+    resolve_fallbacks,
+    without_global_fallbacks,
 )
 from .litellm_core_utils.prompt_templates.common_utils import (
     add_system_prompt_to_messages,
@@ -748,7 +750,6 @@ async def acompletion(
         - The `completion` function is called using `run_in_executor` to execute synchronously in the event loop.
         - If `stream` is True, the function returns an async generator that yields completion lines.
     """
-    fallbacks: Final = kwargs.get("fallbacks", None) or litellm.model_fallbacks
     mock_timeout = kwargs.get("mock_timeout", None)
 
     if mock_timeout is True:
@@ -765,6 +766,7 @@ async def acompletion(
     custom_llm_provider = kwargs.get("custom_llm_provider", None)
     _ = _resolve_control_options(kwargs, model)
     caller_kwargs: Final = cast("Mapping[str, object]", {**kwargs})  # cast-ok: acompletion takes untyped kwargs
+    fallbacks: Final = resolve_fallbacks(caller_kwargs)
 
     ## PROMPT MANAGEMENT HOOKS ##
     #########################################################
@@ -922,7 +924,7 @@ async def acompletion(
 
     if fallbacks is not None:
         response = await async_completion_with_fallbacks(
-            **completion_kwargs, kwargs={"fallbacks": fallbacks, **caller_kwargs}
+            **completion_kwargs, kwargs={**caller_kwargs, "fallbacks": fallbacks}
         )
         if response is None:
             raise Exception(
@@ -5784,7 +5786,6 @@ def completion(
     metadata: Final = kwargs.get("metadata", None)
     model_info: Final = kwargs.get("model_info", None)
     proxy_server_request: Final = kwargs.get("proxy_server_request", None)
-    fallbacks = kwargs.get("fallbacks", None)
     provider_specific_header: Final = cast(
         ProviderSpecificHeader | Sequence[ProviderSpecificHeader] | None,
         kwargs.get("provider_specific_header", None),
@@ -5917,10 +5918,14 @@ def completion(
         elif num_retries is not None:
             max_retries = num_retries
         logging: Final[LiteLLMLoggingObj] = cast(LiteLLMLoggingObj, litellm_logging_obj)
-        fallbacks = fallbacks or litellm.model_fallbacks
+        routing_kwargs: Final = cast("Mapping[str, object]", kwargs)  # cast-ok: kwargs is the untyped request dict
+        fallbacks: Final = resolve_fallbacks(routing_kwargs)
         if fallbacks is not None:
-            return completion_with_fallbacks(  # pyright: ignore[reportReturnType]  # fallback runner is untyped; resolves to ModelResponse|CustomStreamWrapper at runtime
-                **args
+            fallback_runner: Final = (
+                async_completion_with_fallbacks if acompletion is True else completion_with_fallbacks
+            )
+            return fallback_runner(  # pyright: ignore[reportReturnType]  # fallback runner is untyped; resolves to ModelResponse|CustomStreamWrapper (or its coroutine) at runtime
+                **{**args, "kwargs": {**kwargs, "fallbacks": fallbacks}}
             )
         if model_list is not None:
             deployments: Final = [m["litellm_params"] for m in model_list if m["model_name"] == model]
@@ -5940,7 +5945,6 @@ def completion(
         if deployment_id is not None:  # azure llms
             model = deployment_id
             custom_llm_provider = "azure"
-        routing_kwargs: Final = cast("Mapping[str, object]", kwargs)  # cast-ok: kwargs is the untyped request dict
         requested_model: Final = model
         model, custom_llm_provider, dynamic_api_key, api_base = get_llm_provider(
             model=model,
@@ -9619,12 +9623,13 @@ async def ahealth_check(
             "batch": ("alist_batches" if custom_llm_provider in LIST_BATCHES_SUPPORTED_PROVIDERS else "acompletion"),
         }.get("decisions" if is_decisions_model_mode(mode) else mode, "acompletion")
         if "*" in model:
-            return await HealthCheckHelpers.ahealth_check_wildcard_models(
-                model=model,
-                custom_llm_provider=custom_llm_provider,
-                model_params=model_params,
-                litellm_logging_obj=litellm_logging_obj,
-            )
+            with without_global_fallbacks():
+                return await HealthCheckHelpers.ahealth_check_wildcard_models(
+                    model=model,
+                    custom_llm_provider=custom_llm_provider,
+                    model_params=model_params,
+                    litellm_logging_obj=litellm_logging_obj,
+                )
 
         mode_handlers: Final = HealthCheckHelpers.get_mode_handlers(
             model=model,
@@ -9636,7 +9641,8 @@ async def ahealth_check(
 
         handler_mode: Final = "decisions" if is_decisions_model_mode(mode) else mode
         if handler_mode in mode_handlers:
-            _response: Final = await mode_handlers[handler_mode]()
+            with without_global_fallbacks():
+                _response: Final = await mode_handlers[handler_mode]()
             _response_headers: Final = cast(  # cast-ok: provider headers are stored as a string-keyed mapping
                 Mapping[str, object], (get_hidden_params(_response) or {}).get("headers", {}) or {}
             )

litellm/proxy/auth/fallback_model_access.py

@@ -9,6 +9,7 @@ checks for each fallback target before it is attempted. Enabled by default. Opt
 key, team and user search tool grants instead, the same check /search runs on the requested tool.
 """
 
+import inspect
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from typing import Final
@@ -67,7 +68,8 @@ async def is_search_tool_authorized_for_token(*, search_tool_name: str, valid_to
 
 
 def _is_search_request(request_kwargs: Mapping[str, object]) -> bool:
-    return request_kwargs.get("original_generic_function") is asearch
+    handler: Final = request_kwargs.get("original_generic_function")
+    return callable(handler) and inspect.unwrap(handler, stop=lambda function: function is asearch) is asearch
 
 
 def _token_in_metadata(metadata: object) -> UserAPIKeyAuth | None:

litellm/proxy/proxy_server.py

@@ -7428,6 +7428,10 @@ class ProxyConfig:
                     litellm.drop_params = drop_params_flag(value, "litellm_settings.drop_params", verbose_proxy_logger)
                 elif key == "suppress_startup_output":
                     continue
+                elif key == "model_fallbacks":
+                    verbose_proxy_logger.warning(
+                        "litellm_settings.model_fallbacks has no effect on the proxy, set router_settings.fallbacks instead"
+                    )
                 else:
                     verbose_proxy_logger.debug(
                         "%s setting litellm.%s=%s%s",

litellm/proxy/route_llm_request.py

@@ -15,6 +15,7 @@ from pydantic import BeforeValidator, TypeAdapter, ValidationError
 
 import litellm
 from litellm.litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params
+from litellm.litellm_core_utils.fallback_utils import without_global_fallbacks_around
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.routing_common_utils import is_proxy_admin_request
 from litellm.litellm_core_utils.total_deadline import enforce_total_timeout_scope
@@ -898,6 +899,9 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
             }
         )
 
+    def direct_litellm_endpoint():  # noqa: ANN202  # the endpoint is looked up by name, so its call stays as untyped as before
+        return without_global_fallbacks_around(getattr(litellm, f"{route_type}"))
+
     team_id: Final = get_team_id_from_data(data)
     router_model_names: Final = llm_router.registry().names(kind="group_set") if llm_router is not None else []
     is_proxy_admin_without_team: Final = team_id is None and is_proxy_admin_request(data)
@@ -911,7 +915,7 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
         if llm_router is not None:
             return getattr(llm_router, f"{route_type}")(**data)
         else:
-            return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs())
+            return direct_litellm_endpoint()(**direct_litellm_kwargs())
 
     elif (
         route_type == "acompletion"
@@ -966,7 +970,7 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
                     # If we can't get deployment creds, continue without them
                     pass
 
-            return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs())
+            return direct_litellm_endpoint()(**direct_litellm_kwargs())
         # Skip model-based routing for container operations
         if route_type in [
             "acreate_container",
@@ -1006,7 +1010,7 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
             else await _authorized_team_file_deployment(llm_router, model_less_team_file_call)
         )
         if team_file_deployment is not None:
-            return getattr(litellm, f"{route_type}")(
+            return direct_litellm_endpoint()(
                 **_with_team_deployment_attribution(
                     {**direct_litellm_kwargs(), **team_file_deployment.credentials},
                     team_file_deployment,
@@ -1034,7 +1038,7 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
             "aingest",
         ] and (data.get("model") is None or data.get("model") == ""):
             # These endpoints don't need a model, use custom_llm_provider directly
-            return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs())
+            return direct_litellm_endpoint()(**direct_litellm_kwargs())
 
         if "model" not in data:
             raise ProxyMissingRequiredParamError(
@@ -1075,7 +1079,7 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
             # Check wildcards before checking deployment_names
             # Priority: 1. Exact model_name match, 2. Wildcard match, 3. deployment_names match
             if llm_router.settings().router_general_settings.pass_through_all_models:
-                return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs())
+                return direct_litellm_endpoint()(**direct_litellm_kwargs())
             elif (
                 llm_router.settings().default_deployment is not None
                 or len(llm_router.registry().patterns().patterns) > 0
@@ -1130,7 +1134,7 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
                     return getattr(llm_router, f"{route_type}")(**data)
                 except Exception:
                     # If router fails (e.g., model not found in router), fall back to direct call
-                    return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs())
+                    return direct_litellm_endpoint()(**direct_litellm_kwargs())
             elif _is_a2a_agent_model(data.get("model", "")):
                 from litellm.proxy.agent_endpoints.a2a_routing import (
                     route_a2a_agent_request,
@@ -1142,7 +1146,7 @@ async def _route_request_single_attempt(  # noqa: ANN202  # returns unawaited pr
                 # Fall through to raise exception below if result is None
 
     elif user_model is not None or route_type == "allm_passthrough_route":
-        return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs())
+        return direct_litellm_endpoint()(**direct_litellm_kwargs())
 
     # if no route found then it's a bad request
     route_name: Final = ROUTE_ENDPOINT_MAPPING.get(route_type, route_type)

litellm/router_python_host.py

@@ -23,6 +23,7 @@ from pydantic import TypeAdapter
 from typing_extensions import overload
 
 import litellm
+from litellm.litellm_core_utils.fallback_utils import without_global_fallbacks_around
 from litellm.router_config import RouterConfig, RouterDynamicConfig, live_dynamic_config
 
 if TYPE_CHECKING:
@@ -111,7 +112,7 @@ class PythonHost:
             raise AttributeError(f"litellm has no endpoint {op!r}")
         if not callable(endpoint):
             raise TypeError(f"litellm.{op} is not an endpoint")
-        return endpoint
+        return without_global_fallbacks_around(endpoint)
 
     def sleep(self, seconds: float) -> Awaitable[None]:
         return asyncio.sleep(seconds)

litellm/utils.py

@@ -106,6 +106,12 @@ from litellm.litellm_core_utils.fallback_generalizations import (
     match_capability_generalizations,
     match_fill_missing_generalizations,
 )
+from litellm.litellm_core_utils.fallback_utils import (
+    checked_fallback_attempt,
+    report_failed_sync_fallback_attempt,
+    uses_fallbacks,
+    without_global_fallbacks,
+)
 from litellm.litellm_core_utils.get_model_cost_map import (
     GetModelCostMap,
     fetch_registration_document,
@@ -2360,7 +2366,11 @@ def client(original_function):
 
             clamp_max_tokens(model, args, kwargs, call_type)
             # MODEL CALL
-            result = original_function(*args, **kwargs)
+            try:
+                result = original_function(*args, **kwargs)
+            except Exception as deployment_error:
+                report_failed_sync_fallback_attempt(kwargs, call_type, deployment_error)
+                raise
             end_time = datetime.datetime.now()
             if _returns_requested_stream(kwargs, call_type, result) or _is_converted_stream_result(result):
                 if "complete_response" in kwargs and kwargs["complete_response"] is True:
@@ -2414,17 +2424,18 @@ def client(original_function):
                 original_function=original_function,
                 rules_obj=rules_obj,
             )
+            checked_result: Final = checked_fallback_attempt(kwargs, call_type, result)
 
             # [OPTIONAL] ADD TO CACHE
             _llm_caching_handler.sync_set_cache(
-                result=result,
+                result=checked_result,
                 args=args,
                 kwargs=kwargs,
             )
 
             _update_response_metadata: Final[_ResponseMetadataUpdater] = litellm_utils.update_response_metadata
             _update_response_metadata(
-                result=result,
+                result=checked_result,
                 logging_obj=logging_obj,
                 model=model,
                 kwargs=kwargs,
@@ -2434,7 +2445,8 @@ def client(original_function):
 
             # LOG SUCCESS - handle streaming success logging in the _next_ object, remove `handle_success` once it's deprecated
             verbose_logger.info("Wrapper: Completed Call, calling success_handler")
-            if not is_internal_call.get():
+            logged_by_fallback_attempt: Final = call_type == CallTypes.completion.value and uses_fallbacks(kwargs)
+            if not is_internal_call.get() and not logged_by_fallback_attempt:
                 # Copy the current context to propagate it to the background thread
                 # This is essential for OpenTelemetry span context propagation
                 ctx: Final = contextvars.copy_context()
@@ -2442,12 +2454,12 @@ def client(original_function):
                 executor.submit(
                     ctx.run,
                     logging_obj.success_handler,
-                    result,
+                    checked_result,
                     start_time,
                     end_time,
                 )
             # RETURN RESULT
-            return result
+            return checked_result
         except Exception as e:
             if call_type == CallTypes.completion.value:
                 num_retries = kwargs.get("num_retries", None) or litellm.num_retries or None
@@ -2546,7 +2558,7 @@ def client(original_function):
             kwargs["litellm_call_id"] = str(uuid.uuid4())
 
         model: Final[str | None] = args[0] if len(args) > 0 else kwargs.get("model", None)
-        is_completion_with_fallbacks: Final = kwargs.get("fallbacks") is not None
+        is_completion_with_fallbacks: Final = bool(kwargs.get("fallbacks"))
         kwargs.pop("_is_litellm_internal_call", None)  # discard if injected
         _is_litellm_internal_call: Final = is_internal_call.get()
         _deployment_call_end_time: datetime.datetime | None = None
@@ -8501,7 +8513,8 @@ def check_valid_key(model: str, api_key: str):
     """
     messages: Final = [{"role": "user", "content": "Hey, how's it going?"}]
     try:
-        litellm.completion(model=model, messages=messages, api_key=api_key, max_tokens=10)
+        with without_global_fallbacks():
+            litellm.completion(model=model, messages=messages, api_key=api_key, max_tokens=10)
         return True
     except AuthenticationError:
         return False

tests/unit/litellm_core_utils/test_fallback_utils.py

@@ -1,23 +1,39 @@
 """Tests for litellm.litellm_core_utils.fallback_utils."""
 
+import asyncio
+import contextlib
 import json
 import pathlib
-from typing import Final
+import threading
+from collections.abc import Mapping
+from concurrent.futures import ThreadPoolExecutor
+from typing import Final, NoReturn
 
 import httpx
 import pytest
 import respx
+from pydantic import TypeAdapter
 
 import litellm
+from litellm.caching.caching import Cache, DualCache
+from litellm.integrations.custom_guardrail import CustomGuardrail
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils import litellm_logging
 from litellm.litellm_core_utils.core_helpers import process_response_headers
 from litellm.litellm_core_utils.fallback_utils import (
     async_completion_with_fallbacks,
 )
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
+from litellm.llms.chatgpt.common_utils import CHATGPT_DEVICE_CODE_URL
 from litellm.main import acompletion as python_acompletion
 from litellm.main import completion as python_completion
+from litellm.types.guardrails import GuardrailEventHooks
+from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
+from litellm.types.utils import CallTypes, CallTypesLiteral, Choices, CredentialItem, LLMResponseTypes, Message, ModelResponse
 from litellm.vector_stores.vector_store_registry import LiteLLM_ManagedVectorStore, VectorStoreRegistry
 
+LITELLM_PARAMS: Final = TypeAdapter(dict[str, object])
+
 
 @pytest.mark.asyncio
 async def test_fallback_dict_not_mutated(monkeypatch):
@@ -179,9 +195,7 @@ def test_process_response_headers_ignores_preserve_flag_for_httpx_headers():
 
 
 @pytest.mark.asyncio
-async def test_acompletion_fallbacks_bad_models(
-    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
-) -> None:
+async def test_acompletion_fallbacks_bad_models(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
     route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
         side_effect=(
@@ -210,7 +224,7 @@ async def test_acompletion_fallbacks_bad_models(
         )
     )
 
-    with pytest.raises(Exception, match="All fallback attempts failed"):
+    with pytest.raises(Exception, match=r"fallback unavailable\. All fallback attempts failed"):
         await python_acompletion(
             model="openai/first",
             messages=[{"role": "user", "content": "hello"}],
@@ -222,9 +236,7 @@ async def test_acompletion_fallbacks_bad_models(
 
 
 @pytest.mark.asyncio
-async def test_acompletion_fallbacks_basic(
-    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
-) -> None:
+async def test_acompletion_fallbacks_basic(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
     route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
         side_effect=(
@@ -271,9 +283,7 @@ async def test_acompletion_fallbacks_basic(
 
 
 @pytest.mark.asyncio
-async def test_acompletion_fallbacks_empty_list(
-    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
-) -> None:
+async def test_acompletion_fallbacks_empty_list(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
     route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
         return_value=httpx.Response(
@@ -401,9 +411,7 @@ async def test_acompletion_fallbacks_with_dict_config(
     )
 
 
-def test_completion_fallbacks_sync(
-    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
-) -> None:
+def test_completion_fallbacks_sync(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
     route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
         side_effect=(
@@ -449,6 +457,27 @@ def test_completion_fallbacks_sync(
     assert tuple(json.loads(call.request.content)["model"] for call in route.calls) == ("first", "second")
 
 
+@pytest.mark.timeout(30, method="thread")
+def test_completion_fallbacks_name_the_last_error_when_every_model_fails(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    fallback_unavailable: Final = {
+        "error": {"message": "fallback unavailable", "type": "invalid_request_error", "param": None, "code": None}
+    }
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=(_model_unavailable(), httpx.Response(404, json=fallback_unavailable))
+    )
+
+    with pytest.raises(Exception, match=r"fallback unavailable\. All fallback attempts failed"):
+        python_completion(
+            model="openai/first",
+            messages=[{"role": "user", "content": "hello"}],
+            api_key="test-key",
+            fallbacks=["openai/second"],
+        )
+
+
 CACHE_POINTS: Final = ({"location": "message", "role": "system"},)
 CACHED_MESSAGES: Final = (
     {"role": "system", "content": "You are a careful assistant."},
@@ -631,3 +660,763 @@ async def test_acompletion_fallback_sends_the_prompt_the_primary_model_prefix_te
     direct_body, fallback_body = (json.loads(call.request.content) for call in anthropic_route.calls)
     assert "What color is the sky?" in json.dumps(direct_body)
     assert fallback_body == direct_body
+
+
+def _model_unavailable() -> httpx.Response:
+    return httpx.Response(
+        404,
+        json={
+            "error": {
+                "message": "model unavailable",
+                "type": "invalid_request_error",
+                "param": None,
+                "code": "model_not_found",
+            }
+        },
+    )
+
+
+def _replies(contents: tuple[str | None, ...]) -> tuple[httpx.Response, ...]:
+    return tuple(
+        _model_unavailable()
+        if content is None
+        else httpx.Response(
+            200,
+            json={
+                **OPENAI_ANSWER,
+                "choices": [
+                    {"index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop"}
+                ],
+            },
+        )
+        for content in contents
+    )
+
+
+def _requested_models(route: respx.Route) -> tuple[str, ...]:
+    return tuple(json.loads(call.request.content)["model"] for call in route.calls)
+
+
+GLOBAL_CHAIN_OUTCOMES: Final = (
+    pytest.param(("first answer",), ("first",), id="primary-answers"),
+    pytest.param((None, None, "third answer"), ("first", "second", "third"), id="falls-through-in-order"),
+)
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.parametrize(("contents", "expected_models"), GLOBAL_CHAIN_OUTCOMES)
+@pytest.mark.asyncio
+async def test_acompletion_tries_the_global_model_fallbacks_once_each_in_order(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    contents: tuple[str | None, ...],
+    expected_models: tuple[str, ...],
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second", "openai/third"])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(contents))
+
+    response: Final = await python_acompletion(
+        model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key"
+    )
+
+    assert response.choices[0].message.content == contents[-1]
+    assert _requested_models(route) == expected_models
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.parametrize(("contents", "expected_models"), GLOBAL_CHAIN_OUTCOMES)
+def test_completion_tries_the_global_model_fallbacks_once_each_in_order(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    contents: tuple[str | None, ...],
+    expected_models: tuple[str, ...],
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second", "openai/third"])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(contents))
+
+    response: Final = python_completion(
+        model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key"
+    )
+
+    assert response.choices[0].message.content == contents[-1]
+    assert _requested_models(route) == expected_models
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+async def test_acompletion_per_call_fallbacks_win_over_the_global_model_fallbacks(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/global-fallback"])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies((None, "per-call answer"))
+    )
+
+    response: Final = await python_acompletion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "hello"}],
+        api_key="test-key",
+        fallbacks=["openai/second"],
+    )
+
+    assert response.choices[0].message.content == "per-call answer"
+    assert _requested_models(route) == ("first", "second")
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+async def test_acompletion_empty_per_call_fallbacks_opt_out_of_the_global_model_fallbacks(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies((None, "fallback answer"))
+    )
+
+    with pytest.raises(litellm.NotFoundError, match="model unavailable"):
+        await python_acompletion(
+            model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key", fallbacks=[]
+        )
+
+    assert _requested_models(route) == ("first",)
+
+
+async def _flush_success_logs() -> None:
+    await asyncio.sleep(0)
+    GLOBAL_LOGGING_WORKER.start()
+    await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10.0)
+
+
+class _SuccessRecorder(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.models: list[object] = []
+        self.call_ids: list[object] = []
+
+    async def async_log_success_event(
+        self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self.models.append(kwargs.get("model"))
+        self.call_ids.append(LITELLM_PARAMS.validate_python(kwargs.get("litellm_params")).get("litellm_call_id"))
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+async def test_acompletion_through_the_global_model_fallbacks_logs_one_success(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    recorder: Final = _SuccessRecorder()
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"])
+    monkeypatch.setattr(litellm, "callbacks", [recorder])
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies((None, "fallback answer"))
+    )
+
+    await python_acompletion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "hello"}],
+        api_key="test-key",
+        litellm_call_id="caller-call-id",
+    )
+    await _flush_success_logs()
+
+    assert recorder.models == ["second"]
+    assert recorder.call_ids == ["caller-call-id"]
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.parametrize(
+    "global_fallbacks",
+    (pytest.param(["openai/second"], id="global-set"), pytest.param(None, id="global-unset")),
+)
+@pytest.mark.asyncio
+async def test_acompletion_with_empty_per_call_fallbacks_logs_one_success(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, global_fallbacks: list[str] | None
+) -> None:
+    recorder: Final = _SuccessRecorder()
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", global_fallbacks)
+    monkeypatch.setattr(litellm, "callbacks", [recorder])
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(("first answer",)))
+
+    await python_acompletion(
+        model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key", fallbacks=[]
+    )
+    await _flush_success_logs()
+
+    assert recorder.models == ["first"]
+
+
+class _EverySuccessRecorder(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.events: list[tuple[str, object]] = []
+
+    def log_success_event(
+        self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self.events.append(("sync", kwargs.get("model")))
+
+    async def async_log_success_event(
+        self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self.events.append(("async", kwargs.get("model")))
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+async def test_atext_completion_through_the_global_model_fallbacks_logs_one_success(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    recorder: Final = _EverySuccessRecorder()
+    logging_executor: Final = ThreadPoolExecutor(max_workers=2)
+    monkeypatch.setattr(litellm.utils, "executor", logging_executor)
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"])
+    monkeypatch.setattr(litellm, "callbacks", [recorder])
+    monkeypatch.setenv("OPENAI_API_KEY", "test-key")
+    chat_route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies(("fallback answer",))
+    )
+    text_route: Final = respx_mock.post("https://api.openai.com/v1/completions").mock(
+        side_effect=(_model_unavailable(),)
+    )
+
+    response: Final = await litellm.atext_completion(model="openai/first", prompt="hello")
+    await _flush_success_logs()
+    logging_executor.shutdown(wait=True)
+
+    assert response.choices[0].text == "fallback answer"
+    assert _requested_models(text_route) == ("first",)
+    assert _requested_models(chat_route) == ("second",)
+    assert len(recorder.events) == 1
+
+
+class _DeploymentAttemptRecorder(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.attempts: list[tuple[str, object]] = []
+
+    async def async_post_call_failure_deployment_hook(
+        self,
+        request_data: Mapping[str, object],
+        exception: Exception,
+        call_type: CallTypes | None,
+        fallback_depth: int | None = None,
+    ) -> None:
+        self.attempts.append(("failure", request_data.get("model")))
+
+    async def async_post_call_success_deployment_hook(
+        self, request_data: dict[str, object], response: LLMResponseTypes, call_type: CallTypes | None
+    ) -> LLMResponseTypes | None:
+        self.attempts.append(("success", request_data.get("model")))
+        return response
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_with_fallbacks_reports_every_attempt_to_the_deployment_hooks_in_order(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    recorder: Final = _DeploymentAttemptRecorder()
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "callbacks", [recorder])
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies((None, None, "third answer"))
+    )
+
+    response: Final = python_completion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "hello"}],
+        api_key="test-key",
+        fallbacks=["openai/second", "openai/third"],
+    )
+
+    assert response.choices[0].message.content == "third answer"
+    assert recorder.attempts == [("failure", "openai/first"), ("failure", "openai/second"), ("success", "openai/third")]
+
+
+CREDENTIAL_ROUTES: Final = (
+    pytest.param(
+        "openai/second",
+        "https://gateway.example/v1/chat/completions",
+        ("authorization", "Bearer caller-key"),
+        id="same-provider-keeps-the-callers-credentials",
+    ),
+    pytest.param(
+        "anthropic/claude-haiku-5-5",
+        "https://api.anthropic.com/v1/messages",
+        ("x-api-key", "anthropic-env-key"),
+        id="other-provider-uses-its-own-credentials",
+    ),
+)
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.parametrize(("fallback", "fallback_url", "credential"), CREDENTIAL_ROUTES)
+@pytest.mark.asyncio
+async def test_acompletion_fallback_sends_credentials_only_to_the_provider_they_belong_to(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    fallback: str,
+    fallback_url: str,
+    credential: tuple[str, str],
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", [fallback])
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key")
+    _ = respx_mock.post("https://gateway.example/v1/chat/completions").mock(
+        side_effect=(_model_unavailable(), httpx.Response(200, json=OPENAI_ANSWER))
+    )
+    _ = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
+        return_value=httpx.Response(200, json=ANTHROPIC_ANSWER)
+    )
+
+    response: Final = await python_acompletion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "hello"}],
+        api_key="caller-key",
+        api_base="https://gateway.example/v1",
+    )
+
+    header, value = credential
+    assert response.choices[0].message.content == "ok"
+    assert str(respx_mock.calls.last.request.url) == fallback_url
+    assert respx_mock.calls.last.request.headers[header] == value
+
+
+GATEWAY_CREDENTIAL_CARRIERS: Final = (
+    pytest.param("openai/first", {"litellm_credential_name": "gateway"}, id="named-credential"),
+    pytest.param(
+        "openai/first",
+        {
+            "api_key": "gateway-key",
+            "api_base": "https://gateway.example/v1",
+            "headers": {"authorization": "Bearer gateway-key"},
+        },
+        id="request-headers",
+    ),
+    pytest.param(
+        "first-custom-model",
+        {"custom_llm_provider": "openai", "api_key": "gateway-key", "api_base": "https://gateway.example/v1"},
+        id="explicit-provider",
+    ),
+)
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.parametrize(("model", "gateway_credentials"), GATEWAY_CREDENTIAL_CARRIERS)
+@pytest.mark.asyncio
+async def test_acompletion_fallback_to_another_provider_never_reuses_the_first_providers_credentials(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    model: str,
+    gateway_credentials: dict[str, object],
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(
+        litellm,
+        "credential_list",
+        [
+            CredentialItem(
+                credential_name="gateway",
+                credential_values={"api_key": "gateway-key", "api_base": "https://gateway.example/v1"},
+                credential_info={},
+            )
+        ],
+    )
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key")
+    _ = respx_mock.post("https://gateway.example/v1/chat/completions").mock(return_value=_model_unavailable())
+    _ = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
+        return_value=httpx.Response(200, json=ANTHROPIC_ANSWER)
+    )
+
+    response: Final = await python_acompletion(
+        model=model,
+        messages=[{"role": "user", "content": "hello"}],
+        fallbacks=["anthropic/claude-haiku-5-5"],
+        **gateway_credentials,
+    )
+
+    fallback_request: Final = respx_mock.calls.last.request
+    assert response.choices[0].message.content == "ok"
+    assert str(fallback_request.url) == "https://api.anthropic.com/v1/messages"
+    assert fallback_request.headers["x-api-key"] == "anthropic-env-key"
+    assert not any("gateway-key" in value for value in fallback_request.headers.values())
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+async def test_acompletion_fallback_to_an_unprefixed_model_keeps_the_requests_provider_and_credentials(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    route: Final = respx_mock.post("https://models.example/v1/chat/completions").mock(
+        side_effect=_replies((None, "fallback answer"))
+    )
+
+    response: Final = await python_acompletion(
+        model="first-custom-model",
+        custom_llm_provider="openai",
+        api_base="https://models.example/v1",
+        api_key="models-key",
+        messages=[{"role": "user", "content": "hello"}],
+        fallbacks=["second-custom-model"],
+    )
+
+    assert response.choices[0].message.content == "fallback answer"
+    assert _requested_models(route) == ("first-custom-model", "second-custom-model")
+    assert route.calls.last.request.headers["authorization"] == "Bearer models-key"
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_fallback_from_a_model_without_a_provider_never_reuses_its_credentials(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key")
+    route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
+        return_value=httpx.Response(200, json=ANTHROPIC_ANSWER)
+    )
+
+    response: Final = python_completion(
+        model="first-without-a-provider",
+        messages=[{"role": "user", "content": "hello"}],
+        api_key="first-key",
+        fallbacks=["anthropic/claude-haiku-5-5"],
+    )
+
+    assert response.choices[0].message.content == "ok"
+    assert route.calls.last.request.headers["x-api-key"] == "anthropic-env-key"
+
+
+async def _router_completion(router: litellm.Router, *, is_async: bool) -> object:
+    messages: Final = [{"role": "user", "content": "hello"}]
+    if is_async:
+        return await router.acompletion(model="first", messages=messages)
+    return router.completion(model="first", messages=messages)
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.parametrize("is_async", [True, False])
+@pytest.mark.asyncio
+async def test_router_leaves_fallbacks_to_the_router_when_global_model_fallbacks_are_set(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, is_async: bool
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies((None, "global fallback answer"))
+    )
+    router: Final = litellm.Router(
+        model_list=[{"model_name": "first", "litellm_params": {"model": "openai/first", "api_key": "test-key"}}],
+        num_retries=0,
+    )
+
+    with pytest.raises(litellm.NotFoundError, match="model unavailable"):
+        await _router_completion(router, is_async=is_async)
+
+    assert _requested_models(route) == ("first",)
+
+
+STREAMED_ANSWER: Final = (
+    "".join(
+        f"data: {json.dumps(chunk)}\n\n"
+        for chunk in (
+            {
+                "id": "chatcmpl-1",
+                "object": "chat.completion.chunk",
+                "created": 1,
+                "model": "first",
+                "choices": [{"index": 0, "delta": {"role": "assistant", "content": "streamed answer"}}],
+            },
+            {
+                "id": "chatcmpl-1",
+                "object": "chat.completion.chunk",
+                "created": 1,
+                "model": "first",
+                "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
+            },
+        )
+    )
+    + "data: [DONE]\n\n"
+)
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_streams_through_the_global_model_fallbacks(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"])
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        return_value=httpx.Response(200, text=STREAMED_ANSWER, headers={"content-type": "text/event-stream"})
+    )
+
+    stream: Final = python_completion(
+        model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key", stream=True
+    )
+
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in stream) == "streamed answer"
+
+
+class _ConcurrentSyncSuccessRecorder(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.models: list[object] = []
+        self._pair_concurrent_logs: Final = threading.Barrier(2, timeout=0.5)
+
+    def logging_hook(
+        self, kwargs: dict[str, object], result: object, call_type: str
+    ) -> tuple[dict[str, object], object]:
+        with contextlib.suppress(threading.BrokenBarrierError):
+            _ = self._pair_concurrent_logs.wait()
+        return kwargs, result
+
+    def log_success_event(
+        self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self.models.append(kwargs.get("model"))
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_through_the_global_model_fallbacks_logs_one_success(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    recorder: Final = _ConcurrentSyncSuccessRecorder()
+    logging_executor: Final = ThreadPoolExecutor(max_workers=2)
+    monkeypatch.setattr(litellm.utils, "executor", logging_executor)
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"])
+    monkeypatch.setattr(litellm, "callbacks", [recorder])
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies((None, "fallback answer"))
+    )
+
+    python_completion(model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key")
+    logging_executor.shutdown(wait=True)
+
+    assert recorder.models == ["second"]
+
+
+class _RedactingGuardrail(CustomGuardrail):
+    def __init__(self) -> None:
+        super().__init__(
+            guardrail_name="redact", event_hook=[GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call]
+        )
+
+    async def async_pre_call_hook(
+        self, user_api_key_dict: UserAPIKeyAuth, cache: DualCache, data: dict, call_type: CallTypesLiteral
+    ) -> dict[str, object]:
+        return {**data, "messages": [{"role": "user", "content": "[REDACTED]"}]}
+
+    async def async_post_call_success_hook(
+        self, data: dict, user_api_key_dict: UserAPIKeyAuth, response: LLMResponseTypes
+    ) -> ModelResponse:
+        return ModelResponse(model="checked", choices=[Choices(message=Message(content="[CHECKED]"))])
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_with_fallbacks_runs_the_requested_guardrail_before_every_attempt(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "callbacks", [_RedactingGuardrail()])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies((None, "fallback answer"))
+    )
+
+    response: Final = python_completion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "my card is 4111 1111 1111 1111"}],
+        api_key="test-key",
+        fallbacks=["openai/second"],
+        guardrails=["redact"],
+    )
+
+    assert response.choices[0].message.content == "[CHECKED]"
+    assert [json.loads(call.request.content)["messages"] for call in route.calls] == [
+        [{"role": "user", "content": "[REDACTED]"}]
+    ] * 2
+
+
+class _RejectingGuardrail(CustomGuardrail):
+    def __init__(self) -> None:
+        super().__init__(guardrail_name="reject", event_hook=GuardrailEventHooks.post_call)
+
+    async def async_post_call_success_hook(
+        self, data: dict, user_api_key_dict: UserAPIKeyAuth, response: LLMResponseTypes
+    ) -> NoReturn:
+        raise ValueError("answer rejected")
+
+
+def _completion_checked_by_the_rejecting_guardrail() -> object:
+    return python_completion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "hello"}],
+        api_key="test-key",
+        fallbacks=["openai/second"],
+        guardrails=["reject"],
+    )
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_with_fallbacks_never_caches_an_answer_its_guardrail_rejected(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "cache", Cache(type="local"))
+    monkeypatch.setattr(litellm, "callbacks", [_RejectingGuardrail()])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies(("rejected answer",) * 4)
+    )
+
+    with pytest.raises(Exception, match="answer rejected"):
+        _ = _completion_checked_by_the_rejecting_guardrail()
+    with pytest.raises(Exception, match="answer rejected"):
+        _ = _completion_checked_by_the_rejecting_guardrail()
+
+    assert _requested_models(route) == ("first", "second", "first", "second")
+
+
+class _SyncAnswerRecorder(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.answers: list[object] = []
+
+    def log_success_event(
+        self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self.answers.append(
+            response_obj.choices[0].message.content if isinstance(response_obj, ModelResponse) else None
+        )
+
+
+@pytest.mark.parametrize(
+    ("guardrail_class", "logged_answers"),
+    [(_RedactingGuardrail, ["[CHECKED]"]), (_RejectingGuardrail, [])],
+    ids=["rewritten", "rejected"],
+)
+@pytest.mark.timeout(30, method="thread")
+def test_completion_with_fallbacks_logs_only_the_answer_its_guardrail_returned(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    guardrail_class: type[CustomGuardrail],
+    logged_answers: list[str],
+) -> None:
+    guardrail: Final = guardrail_class()
+    recorder: Final = _SyncAnswerRecorder()
+    logging_executor: Final = ThreadPoolExecutor(max_workers=2)
+    monkeypatch.setattr(litellm.utils, "executor", logging_executor)
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "callbacks", [guardrail, recorder])
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies((None, "raw answer")))
+
+    with contextlib.suppress(Exception):
+        _ = python_completion(
+            model="openai/first",
+            messages=[{"role": "user", "content": "hello"}],
+            api_key="test-key",
+            fallbacks=["openai/second"],
+            guardrails=[guardrail.guardrail_name],
+        )
+    logging_executor.shutdown(wait=True)
+
+    assert recorder.answers == logged_answers
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_streams_through_fallbacks_with_the_requested_guardrail(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "callbacks", [_RedactingGuardrail()])
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        return_value=httpx.Response(200, text=STREAMED_ANSWER, headers={"content-type": "text/event-stream"})
+    )
+
+    stream: Final = python_completion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "my card is 4111 1111 1111 1111"}],
+        api_key="test-key",
+        stream=True,
+        fallbacks=["openai/second"],
+        guardrails=["redact"],
+    )
+
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in stream) == "streamed answer"
+    assert json.loads(route.calls.last.request.content)["messages"] == [{"role": "user", "content": "[REDACTED]"}]
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "primary_price",
+    ({"input_cost_per_token": 1.0, "output_cost_per_token": 1.0}, {"base_model": "gpt-5.4"}),
+)
+async def test_acompletion_fallback_to_another_model_is_priced_like_a_direct_call_to_it(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, primary_price: dict[str, object]
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=(
+            _model_unavailable(),
+            httpx.Response(200, json=OPENAI_ANSWER),
+            httpx.Response(200, json=OPENAI_ANSWER),
+        )
+    )
+    through_fallback: Final = await python_acompletion(
+        model="openai/first",
+        messages=[{"role": "user", "content": "hello"}],
+        api_key="test-key",
+        fallbacks=["openai/gpt-5.4-nano"],
+        **primary_price,
+    )
+    direct: Final = await python_acompletion(
+        model="openai/gpt-5.4-nano", messages=[{"role": "user", "content": "hello"}], api_key="test-key"
+    )
+
+    assert through_fallback._hidden_params["response_cost"] == direct._hidden_params["response_cost"]
+
+
+@pytest.mark.timeout(30, method="thread")
+def test_completion_never_signs_in_to_a_chatgpt_fallback_the_call_does_not_use(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["chatgpt/gpt-5.4"])
+    monkeypatch.setenv("CHATGPT_TOKEN_DIR", str(tmp_path))
+    device_login: Final = respx_mock.post(CHATGPT_DEVICE_CODE_URL).mock(return_value=httpx.Response(503))
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(("first answer",)))
+
+    response: Final = python_completion(
+        model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key"
+    )
+
+    assert response.choices[0].message.content == "first answer"
+    assert not device_login.called
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+async def test_acompletion_falls_back_from_a_chatgpt_model_that_is_not_signed_in(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"])
+    monkeypatch.setenv("CHATGPT_TOKEN_DIR", str(tmp_path))
+    monkeypatch.setenv("OPENAI_API_KEY", "openai-env-key")
+    route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        side_effect=_replies(("fallback answer",))
+    )
+
+    response: Final = await python_acompletion(
+        model="gpt-5.4", custom_llm_provider="chatgpt", messages=[{"role": "user", "content": "hello"}]
+    )
+
+    assert response.choices[0].message.content == "fallback answer"
+    assert _requested_models(route) == ("second",)

tests/unit/litellm_core_utils/test_health_check_helpers.py

@@ -1987,3 +1987,77 @@ async def test_wildcard_health_check_with_pinned_provider_calls_a_model_without_
     assert "error" not in result, result
     assert json.loads(together_route.calls.last.request.content)["model"] == "fake-together-chat"
     assert not other_hosts.called
+
+
+HEALTH_CHECK_ROUTES: Final = (
+    pytest.param(
+        "chat",
+        {"model": "openai/missing-model", "api_base": "https://deployment.example/v1"},
+        "https://deployment.example/v1/chat/completions",
+        id="chat",
+    ),
+    pytest.param(
+        "responses",
+        {"model": "anthropic/missing-model", "api_base": "https://deployment.example"},
+        "https://deployment.example/v1/messages",
+        id="responses-bridged-to-chat",
+    ),
+)
+
+
+def _mock_failing_deployment_with_a_healthy_openai_fallback(
+    respx_mock: respx.MockRouter, deployment_url: str
+) -> respx.Route:
+    _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
+        return_value=httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl-1",
+                "object": "chat.completion",
+                "created": 1,
+                "model": "healthy-fallback",
+                "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
+                "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+            },
+        )
+    )
+    return respx_mock.post(deployment_url).mock(
+        return_value=httpx.Response(404, json={"error": {"message": "model unavailable", "type": "not_found"}})
+    )
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.parametrize(("mode", "deployment_params", "deployment_url"), HEALTH_CHECK_ROUTES)
+@pytest.mark.asyncio
+async def test_ahealth_check_reports_a_failing_deployment_whatever_the_global_model_fallbacks(
+    respx_mock: respx.MockRouter,
+    monkeypatch: pytest.MonkeyPatch,
+    mode: Literal["chat", "responses"],
+    deployment_params: dict[str, str],
+    deployment_url: str,
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/healthy-fallback"])
+    deployment: Final = _mock_failing_deployment_with_a_healthy_openai_fallback(respx_mock, deployment_url)
+
+    result: Final = await ahealth_check({**deployment_params, "api_key": "sk-test"}, mode=mode)
+
+    assert "error" in result
+    assert len(deployment.calls) == 1
+
+
+@pytest.mark.timeout(30, method="thread")
+@pytest.mark.asyncio
+async def test_calls_after_a_health_check_still_use_the_global_model_fallbacks(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/healthy-fallback"])
+    monkeypatch.setenv("OPENAI_API_KEY", "openai-env-key")
+    deployment: Final = {"model": "anthropic/missing-model", "api_base": "https://deployment.example", "api_key": "sk-test"}
+    _ = _mock_failing_deployment_with_a_healthy_openai_fallback(respx_mock, "https://deployment.example/v1/messages")
+
+    _ = await ahealth_check(deployment, mode="chat")
+    response: Final = await litellm.acompletion(**deployment, messages=[{"role": "user", "content": "hi"}])
+
+    assert response.choices[0].message.content == "ok"

tests/unit/proxy/test_proxy_config_unit_test.py

@@ -18,6 +18,9 @@ import io
 import asyncio
 import logging
 
+import yaml
+
+import litellm
 from litellm.proxy.proxy_server import ProxyConfig
 
 INVALID_FILES = ["config_with_missing_include.yaml"]
@@ -355,3 +358,23 @@ async def test_load_config_rejects_unknown_mcp_disabled_capability(tmp_path: Pat
 
     with pytest.raises(ValidationError):
         await ProxyConfig().load_config(router=None, config_file_path=str(config_path))
+
+
+@pytest.mark.asyncio
+async def test_proxy_config_model_fallbacks_never_becomes_the_sdk_global_fallback_list(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(litellm, "model_fallbacks", None)
+    config_file: Final = tmp_path / "config.yaml"
+    _ = config_file.write_text(
+        yaml.safe_dump(
+            {
+                "model_list": [{"model_name": "gpt-5.4-nano", "litellm_params": {"model": "openai/gpt-5.4-nano"}}],
+                "litellm_settings": {"model_fallbacks": ["anthropic/claude-haiku-5-5"]},
+            }
+        )
+    )
+
+    _ = await ProxyConfig().load_config(router=None, config_file_path=str(config_file))
+
+    assert litellm.model_fallbacks is None

tests/unit/proxy/test_route_llm_request.py

@@ -398,7 +398,7 @@ async def test_route_request_no_model_required_with_router_settings_and_no_route
     }
 
     with patch.object(litellm, "acompletion", return_value="fake_response") as mock_completion:
-        await route_request(data, None, "gpt-3.5-turbo", "acompletion")
+        assert await (await route_request(data, None, "gpt-3.5-turbo", "acompletion")) == "fake_response"
 
         mock_completion.assert_called_once_with(**data)
 
@@ -1480,6 +1480,40 @@ async def test_route_request_keeps_key_fallbacks_out_of_pass_through_provider_bo
     assert provider_bodies and all("gpt-4" not in body for body in provider_bodies), provider_bodies
 
 
+@pytest.mark.asyncio
+async def test_route_request_pass_through_never_falls_back_to_the_global_model_fallbacks(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "model_fallbacks", ["openai/gpt-not-granted"])
+    monkeypatch.setenv("OPENAI_API_KEY", "sk-pass-through")
+    monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
+    monkeypatch.delenv("OPENAI_API_BASE", raising=False)
+    router: Final = litellm.Router(
+        model_list=[
+            {
+                "model_name": "gpt-4",
+                "litellm_params": {"model": "openai/gpt-4", "api_key": "sk-proxy", "api_base": _PROXY_DEPLOYMENT_BASE},
+            }
+        ],
+        router_general_settings=RouterGeneralSettings(pass_through_all_models=True),
+    )
+    data: Final = {"model": "openai/gpt-pass-through", "messages": list(_HELLO_MESSAGES)}
+
+    with respx.mock(assert_all_called=False) as respx_mock:
+        requested: Final = respx_mock.post(
+            "https://api.openai.com/v1/chat/completions", json__model="gpt-pass-through"
+        ).mock(return_value=httpx.Response(404, json={"error": {"message": "no such model", "code": "model_not_found"}}))
+        not_granted: Final = respx_mock.post(
+            "https://api.openai.com/v1/chat/completions", json__model="gpt-not-granted"
+        ).mock(return_value=_chat_completion_response("gpt-not-granted"))
+        with pytest.raises(litellm.NotFoundError):
+            await (await route_request(data, router, None, "acompletion"))
+
+    assert requested.called
+    assert not not_granted.called
+
+
 @pytest.mark.parametrize(
     "route_type, param, route",
     [

tests/unit/test_utils.py

@@ -11107,6 +11107,36 @@ def test_check_valid_key_returns_false_for_http_unauthorized() -> None:
         assert upstream.calls.last.request.headers["Authorization"] == "Bearer invalid-test-key"
 
 
+def test_check_valid_key_never_counts_a_global_fallback_answer_as_a_valid_key(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(litellm, "model_fallbacks", ["anthropic/claude-haiku-5-5"])
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key")
+    upstream: Final[respx.MockRouter]
+
+    with respx.mock(assert_all_called=False) as upstream:
+        _ = upstream.post("https://api.openai.com/v1/chat/completions").respond(
+            401, json={"error": {"message": "Invalid API key", "type": "invalid_request_error", "code": "invalid_api_key"}}
+        )
+        fallback_route: Final = upstream.post("https://api.anthropic.com/v1/messages").respond(
+            200,
+            json={
+                "id": "msg_test",
+                "type": "message",
+                "role": "assistant",
+                "model": "claude-haiku-5-5",
+                "content": [{"type": "text", "text": "ok"}],
+                "stop_reason": "end_turn",
+                "usage": {"input_tokens": 1, "output_tokens": 1},
+            },
+        )
+
+        valid_key: Final = check_valid_key(model="gpt-5-mini", api_key="invalid-test-key")
+
+    assert valid_key is False
+    assert not fallback_route.called
+
+
 def test_check_valid_key_returns_true_for_successful_completion() -> None:
     response_body: Final = {
         "id": "chatcmpl-test",