fix: re-check budget on router fallback targets

Budget is enforced once during auth, against the requested model group.
`_is_model_cost_zero` waives every budget check for a zero-cost group, and the
router then picks a fallback target afterwards, inside `run_async_fallback`,
where nothing re-checks budget. A free model with a paid fallback therefore
bills with no budget gate at all.

Add `fallback_budget_check`, the budget sibling of the existing
`fallback_access_check`: a predicate awaited per fallback target that skips
targets the caller cannot pay for. The primary attempt is untouched, so a
zero-cost model is never blocked by budget and only the paid fallback is
refused.

Counter reads pass `max_budget` so `get_current_spend` verifies against
authoritative recorded spend, matching the auth-time key and user checks; a
counter restored from an older snapshot reads as a hit rather than a clean
miss, so without it a stale-low value would keep admitting paid fallbacks.

A zero-cost fallback target is always allowed, and a team key does not inherit
the key owner's personal budget unless `apply_user_budget_to_team_keys` is set,
matching `_PROXY_MaxBudgetLimiter`.

Scope is key and user budgets. Team, team-member, end-user, org, global and
per-model budgets are not covered yet: those auth-path functions enforce rather
than report, so reusing them would fire threshold alerts and take spend
reservations for a target that is then skipped. Two limitations of that scope
are documented in the module docstring: the check reads the spend counter
rather than reserving against it, so concurrent fallbacks can cross a cap
together; and a request reaching the router without
`metadata["user_api_key_auth"]` is not restricted. Both are shared with
`fallback_model_access.py`.

Opt-in via `general_settings.enforce_fallback_budget`.

Relates to #41344

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

litellm/constants.py

@@ -40,6 +40,7 @@ ROUTER_SETTINGS_MANAGED_OUTSIDE_CONFIG: Final[frozenset[str]] = frozenset(
         "router_general_settings",
         "ignore_invalid_deployments",
         "fallback_access_check",
+        "fallback_budget_check",
         "auto_router_capability_limit",
     }
 )

litellm/proxy/auth/fallback_budget.py

@@ -0,0 +1,166 @@
+"""
+Enforce the caller's budget against router fallback targets.
+
+Budget is checked once, during auth, against the *requested* model group. A zero-cost group takes
+`_is_model_cost_zero`'s bypass and waives every budget check; the router then picks a fallback
+target after auth, inside `run_async_fallback`, and nothing re-checks budget on the group that
+actually bills. So a free model with a paid fallback spends without a gate.
+
+This predicate is injected into the router to re-check budget for each fallback target before it is
+attempted, mirroring `fallback_model_access.py`. It deliberately leaves the primary attempt alone:
+a zero-cost model is never blocked by budget, and only the paid fallback is refused. Opt-in via
+`general_settings.enforce_fallback_budget: true`.
+
+Scope: the key's and the user's `max_budget`. Not covered yet, and each needs a read-only evaluation
+path before it can be: team, team-member, end-user, org, global and per-model budgets, whose
+auth-path functions enforce rather than report (they raise), so reusing them would fire threshold
+alerts and take spend reservations for a target that is then skipped; and the key's rolling
+`budget_limits` windows, whose accumulated spend lives only in per-window counters
+(`spend:key:{token}:window:{budget_duration}`), so enforcing them means more counter reads on the
+fallback path rather than reusing state auth already loaded.
+
+Two known limitations of that narrow scope, both shared with `fallback_model_access.py`:
+
+* This reads the spend counter, it does not reserve against it. Requests already in flight all
+  observe the same pre-billing figure, so a cap can be crossed by roughly the number of concurrent
+  fallbacks times their cost. Auth-time enforcement avoids this by pre-filling the counter through
+  `reserve_budget_for_request`, which the zero-cost bypass skips. Turning the soft cap into a hard
+  one means reserving per fallback attempt and reconciling on completion.
+* A request that reaches the router without `metadata["user_api_key_auth"]` is not restricted.
+  Only `add_litellm_data_to_request` populates that key, so endpoints that assemble metadata by
+  hand (for example `/queue/chat/completions`) fall through as unauthenticated.
+"""
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Final
+
+from pydantic import BaseModel, ValidationError
+
+from litellm._logging import verbose_proxy_logger
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.auth.auth_checks import (
+    _is_model_cost_zero,  # pyright: ignore[reportPrivateUsage]  # the zero-cost predicate the auth-time budget checks use; no public equivalent
+)
+from litellm.router import Router
+
+
+class _RequestMetadata(BaseModel):
+    user_api_key_auth: UserAPIKeyAuth | None = None
+
+
+class _FallbackBudgetSettings(BaseModel):
+    enforce_fallback_budget: bool = False
+
+
+def _token_in_metadata(metadata: object) -> UserAPIKeyAuth | None:
+    try:
+        return _RequestMetadata.model_validate(metadata).user_api_key_auth
+    except ValidationError:
+        return None
+
+
+def _user_api_key_auth_from_request(request_kwargs: Mapping[str, object]) -> UserAPIKeyAuth | None:
+    return next(
+        (
+            token
+            for field in ("metadata", "litellm_metadata")
+            if (token := _token_in_metadata(request_kwargs.get(field))) is not None
+        ),
+        None,
+    )
+
+
+def _enforced_by_general_settings() -> bool:
+    from litellm.proxy.proxy_server import general_settings
+
+    return _FallbackBudgetSettings.model_validate(general_settings).enforce_fallback_budget
+
+
+def _applies_user_budget_to_team_keys() -> bool:
+    from litellm.proxy.proxy_server import general_settings
+
+    return general_settings.get("apply_user_budget_to_team_keys") is True
+
+
+async def _counter_spend(counter_key: str, fallback_spend: float, max_budget: float) -> float:
+    """
+    Read a spend counter the same way the auth-time budget checks do.
+
+    `max_budget` is not advisory: it makes `get_current_spend` re-check the counter against the
+    authoritative recorded spend before admitting. A counter restored from an older Redis snapshot
+    reads as a hit rather than a clean miss, so without this the reseed path never runs and a
+    stale-low counter would keep admitting paid fallbacks past the cap.
+    """
+    from litellm.proxy.proxy_server import get_current_spend
+
+    return await get_current_spend(
+        counter_key=counter_key,
+        fallback_spend=fallback_spend,
+        max_budget=max_budget,
+    )
+
+
+async def is_token_within_budget_for_model(*, model: str, valid_token: UserAPIKeyAuth, llm_router: Router) -> bool:
+    """
+    True when the key and the user behind it can still pay for `model`.
+
+    A zero-cost fallback target is always allowed: refusing it would deny a request on spend some
+    other model accrued, which is the same reasoning behind the auth-time bypass.
+    """
+    if _is_model_cost_zero(model=model, llm_router=llm_router):
+        return True
+
+    key_budget: Final = valid_token.max_budget
+    if key_budget is not None and valid_token.token is not None:
+        key_spend: Final = await _counter_spend(
+            counter_key=f"spend:key:{valid_token.token}",
+            fallback_spend=valid_token.spend or 0.0,
+            max_budget=key_budget,
+        )
+        if key_spend >= key_budget:
+            return False
+
+    # Mirrors `_PROXY_MaxBudgetLimiter`: a team key does not carry the key owner's personal budget
+    # unless the proxy opts in, so the personal cap must not gate the fallback either.
+    user_budget: Final = valid_token.user_max_budget
+    if (
+        user_budget is not None
+        and valid_token.user_id is not None
+        and (valid_token.team_id is None or _applies_user_budget_to_team_keys())
+    ):
+        user_spend: Final = await _counter_spend(
+            counter_key=f"spend:user:{valid_token.user_id}",
+            fallback_spend=valid_token.user_spend or 0.0,
+            max_budget=user_budget,
+        )
+        if user_spend >= user_budget:
+            return False
+
+    return True
+
+
+@dataclass(frozen=True, slots=True)
+class RouterFallbackBudgetCheck:
+    """
+    `FallbackBudgetCheck` for the proxy's router: while `is_enforced()` is true, a paid fallback
+    target is attempted only when the caller is still within budget. Requests that carry no key
+    (for example internal health checks) are not restricted.
+    """
+
+    is_enforced: Callable[[], bool]
+
+    async def __call__(self, *, model: str, request_kwargs: Mapping[str, object], llm_router: Router) -> bool:
+        if not self.is_enforced():
+            return True
+        valid_token: Final = _user_api_key_auth_from_request(request_kwargs)
+        if valid_token is None:
+            return True
+        try:
+            return await is_token_within_budget_for_model(model=model, valid_token=valid_token, llm_router=llm_router)
+        except Exception as e:  # noqa: BLE001  # fail closed: a spend lookup failure must not bill the caller
+            verbose_proxy_logger.warning("Skipping fallback to model=%s: budget lookup failed: %s", model, e)
+            return False
+
+
+router_fallback_budget_check: Final = RouterFallbackBudgetCheck(is_enforced=_enforced_by_general_settings)

litellm/proxy/proxy_server.py

@@ -322,6 +322,7 @@ from litellm.proxy.auth.auth_utils import (
     log_once_if_budget_reservation_disabled,
     warn_once_if_custom_auth_skips_common_checks,
 )
+from litellm.proxy.auth.fallback_budget import router_fallback_budget_check
 from litellm.proxy.auth.fallback_model_access import router_fallback_access_check
 from litellm.proxy.auth.handle_jwt import JWTHandler
 from litellm.proxy.auth.litellm_license import AUTO_ROUTER_LICENSE_REMEDY, LicenseCheck
@@ -6153,6 +6154,7 @@ class ProxyConfig:
             ),
             ignore_invalid_deployments=True,  # don't raise an error if a deployment is invalid
             fallback_access_check=router_fallback_access_check,
+            fallback_budget_check=router_fallback_budget_check,
             auto_router_capability_limit=_license_check.auto_router_capability_limit,
         )
 
@@ -6614,6 +6616,7 @@ class ProxyConfig:
                         search_tools=search_tools,
                         ignore_invalid_deployments=True,
                         fallback_access_check=router_fallback_access_check,
+                        fallback_budget_check=router_fallback_budget_check,
                         auto_router_capability_limit=_license_check.auto_router_capability_limit,
                     )
                     verbose_proxy_logger.debug("updated llm_router: %s", llm_router)

litellm/router.py

@@ -240,6 +240,7 @@ from litellm.types.router import (
     DeploymentModelListingInfo,
     DeploymentTypedDict,
     FallbackAccessCheck,
+    FallbackBudgetCheck,
     GuardrailTypedDict,
     LiteLLM_Params,
     MockRouterTestingParams,
@@ -755,6 +756,7 @@ class Router:
         background_health_check_model_groups: Sequence[str] | None = None,
         enable_weighted_failover: bool = False,
         fallback_access_check: FallbackAccessCheck | None = None,
+        fallback_budget_check: FallbackBudgetCheck | None = None,
         auto_router_capability_limit: AutoRouterCapabilityLimit | None = None,
     ) -> None:
         """
@@ -793,6 +795,7 @@ class Router:
             ignore_invalid_deployments (bool): Ignores invalid deployments, and continues with other deployments. Default is to raise an error.
             enable_weighted_failover (bool): When True and the routing strategy is "simple-shuffle", a retryable failure on one deployment causes the request to re-pick (weighted) across the other deployments in the same model group before any cross-group fallback runs. Bounded by `max_fallbacks`. Async-only: currently honored by `router.acompletion()` and other async entrypoints. The sync `router.completion()` path falls back to the regular fallback flow. Defaults to False.
             fallback_access_check (Optional[FallbackAccessCheck]): Awaited before each cross-model-group fallback attempt on the async path; a fallback target it rejects is skipped. Defaults to None (every configured fallback is attempted).
+            fallback_budget_check (Optional[FallbackBudgetCheck]): Awaited before each cross-model-group fallback attempt on the async path; a fallback target it rejects as over budget is skipped. Defaults to None (budget is not re-checked on fallback).
         Returns:
             Router: An instance of the litellm.Router class.
 
@@ -834,6 +837,7 @@ class Router:
         self.ignore_invalid_deployments = ignore_invalid_deployments
         self.auto_router_capability_limit = auto_router_capability_limit
         self.fallback_access_check: Final = fallback_access_check
+        self.fallback_budget_check: Final = fallback_budget_check
         self.debug_level = debug_level
         self.enable_pre_call_checks = enable_pre_call_checks
         self.enable_tag_filtering = enable_tag_filtering

litellm/router_utils/fallback_event_handlers.py

@@ -421,6 +421,25 @@ async def _is_fallback_target_authorized(
     return False
 
 
+async def _is_fallback_target_within_budget(
+    litellm_router: LitellmRouter,
+    fallback_entry: str | Mapping[str, object],
+    original_model_group: str,
+    kwargs: Mapping[str, object],
+) -> bool:
+    budget_check: Final = litellm_router.fallback_budget_check
+    target: Final = _get_fallback_target_model_group(fallback_entry)
+    if budget_check is None or target is None or target == original_model_group:
+        return True
+    if await budget_check(model=target, request_kwargs=kwargs, llm_router=litellm_router):
+        return True
+    verbose_router_logger.info(
+        "Skipping fallback to model_group = %s: caller is over budget",
+        mask_sensitive_structure(fallback_entry),
+    )
+    return False
+
+
 def references_provider_scoped_resource(kwargs: Mapping[str, object]) -> bool:
     """
     True when a file, batch, or fine-tuning job operation names an id that only exists
@@ -528,6 +547,8 @@ async def run_async_fallback(
             continue
         if not await _is_fallback_target_authorized(litellm_router, mg, original_model_group, kwargs):
             continue
+        if not await _is_fallback_target_within_budget(litellm_router, mg, original_model_group, kwargs):
+            continue
         attempt_key = fallback_attempt_key(mg)
         if attempt_key is not None:
             if attempt_key in attempted:

litellm/types/router.py

@@ -963,6 +963,19 @@ class FallbackAccessCheck(Protocol):
     async def __call__(self, *, model: str, request_kwargs: Mapping[str, object], llm_router: "Router") -> bool: ...
 
 
+class FallbackBudgetCheck(Protocol):
+    """
+    Decides whether the caller behind `request_kwargs` is still within budget for fallback `model`.
+
+    Budget is enforced once during auth, against the *requested* model group. A fallback target is
+    chosen later, inside the router, so a zero-cost group that falls back to a priced one bills
+    without any budget gate. The router runs this before every cross-model-group fallback attempt
+    and skips targets it rejects, leaving the free attempt itself untouched.
+    """
+
+    async def __call__(self, *, model: str, request_kwargs: Mapping[str, object], llm_router: "Router") -> bool: ...
+
+
 class AutoRouterCapabilityLimit(Protocol):
     """
     Resolves how many complexity routers may claim each licensed capability right now; None means unlimited.

tests/test_litellm/proxy/auth/test_fallback_budget.py

@@ -0,0 +1,184 @@
+import pytest
+
+from litellm import Router
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.auth.fallback_budget import (
+    RouterFallbackBudgetCheck,
+    is_token_within_budget_for_model,
+)
+
+FREE_MODEL = {
+    "model_name": "free-model",
+    "litellm_params": {
+        "model": "ollama/llama2",
+        "api_base": "http://localhost:11434",
+        "input_cost_per_token": 0.0,
+        "output_cost_per_token": 0.0,
+    },
+    "model_info": {
+        "id": "free-model-id",
+        "input_cost_per_token": 0.0,
+        "output_cost_per_token": 0.0,
+    },
+}
+
+PAID_MODEL = {
+    "model_name": "paid-model",
+    "litellm_params": {"model": "openai/gpt-4o", "api_key": "k"},
+    "model_info": {"id": "paid-model-id"},
+}
+
+
+def _router() -> Router:
+    return Router(model_list=[FREE_MODEL, PAID_MODEL], fallbacks=[{"free-model": ["paid-model"]}])
+
+
+def _token(**overrides) -> UserAPIKeyAuth:
+    fields = {
+        "api_key": "hashed",
+        "token": "hashed",
+        "spend": 0.0,
+        "max_budget": None,
+        "user_id": "u1",
+        "user_spend": 0.0,
+        "user_max_budget": None,
+    }
+    fields.update(overrides)
+    return UserAPIKeyAuth(**fields)
+
+
+ENFORCED = RouterFallbackBudgetCheck(is_enforced=lambda: True)
+NOT_ENFORCED = RouterFallbackBudgetCheck(is_enforced=lambda: False)
+
+
+@pytest.mark.asyncio
+async def test_paid_target_allowed_when_under_budget():
+    token = _token(spend=1.0, max_budget=50.0, user_spend=1.0, user_max_budget=50.0)
+    assert await is_token_within_budget_for_model(model="paid-model", valid_token=token, llm_router=_router()) is True
+
+
+@pytest.mark.asyncio
+async def test_paid_target_refused_when_over_key_budget():
+    token = _token(spend=100.0, max_budget=50.0)
+    assert await is_token_within_budget_for_model(model="paid-model", valid_token=token, llm_router=_router()) is False
+
+
+@pytest.mark.asyncio
+async def test_paid_target_refused_when_over_user_budget():
+    token = _token(user_spend=1900.0, user_max_budget=50.0)
+    assert await is_token_within_budget_for_model(model="paid-model", valid_token=token, llm_router=_router()) is False
+
+
+@pytest.mark.asyncio
+async def test_zero_cost_target_allowed_even_when_over_budget():
+    """Refusing a free target would deny a request on spend some other model accrued."""
+    token = _token(user_spend=1900.0, user_max_budget=50.0)
+    assert await is_token_within_budget_for_model(model="free-model", valid_token=token, llm_router=_router()) is True
+
+
+@pytest.mark.asyncio
+async def test_no_budget_configured_is_always_within_budget():
+    token = _token(spend=9999.0, user_spend=9999.0)
+    assert await is_token_within_budget_for_model(model="paid-model", valid_token=token, llm_router=_router()) is True
+
+
+@pytest.mark.asyncio
+async def test_team_key_does_not_inherit_personal_budget_by_default(monkeypatch):
+    """Mirrors _PROXY_MaxBudgetLimiter: a team key ignores the owner's personal cap."""
+    from litellm.proxy import proxy_server
+
+    monkeypatch.setattr(proxy_server, "general_settings", {}, raising=False)
+    token = _token(team_id="t1", user_spend=1900.0, user_max_budget=50.0)
+    assert await is_token_within_budget_for_model(model="paid-model", valid_token=token, llm_router=_router()) is True
+
+
+@pytest.mark.asyncio
+async def test_team_key_inherits_personal_budget_when_opted_in(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    monkeypatch.setattr(proxy_server, "general_settings", {"apply_user_budget_to_team_keys": True}, raising=False)
+    token = _token(team_id="t1", user_spend=1900.0, user_max_budget=50.0)
+    assert await is_token_within_budget_for_model(model="paid-model", valid_token=token, llm_router=_router()) is False
+
+
+@pytest.mark.asyncio
+async def test_check_is_a_no_op_while_not_enforced():
+    request = {"metadata": {"user_api_key_auth": _token(user_spend=1900.0, user_max_budget=50.0)}}
+    assert await NOT_ENFORCED(model="paid-model", request_kwargs=request, llm_router=_router()) is True
+
+
+@pytest.mark.asyncio
+async def test_request_without_a_key_is_unrestricted():
+    assert await ENFORCED(model="paid-model", request_kwargs={}, llm_router=_router()) is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("metadata_field", ["metadata", "litellm_metadata"])
+async def test_enforced_check_reads_the_key_from_request_metadata(metadata_field: str):
+    over = {metadata_field: {"user_api_key_auth": _token(user_spend=1900.0, user_max_budget=50.0)}}
+    under = {metadata_field: {"user_api_key_auth": _token(user_spend=1.0, user_max_budget=50.0)}}
+
+    assert await ENFORCED(model="paid-model", request_kwargs=over, llm_router=_router()) is False
+    assert await ENFORCED(model="paid-model", request_kwargs=under, llm_router=_router()) is True
+
+
+@pytest.mark.asyncio
+async def test_a_stale_low_counter_still_refuses_a_paid_target(monkeypatch):
+    """
+    The counter can read low (e.g. restored from an older Redis snapshot). Passing the budget makes
+    `get_current_spend` verify against authoritative spend instead of trusting that read, so the
+    paid target is still refused.
+    """
+    from litellm.proxy import proxy_server
+
+    seen: list[dict] = []
+
+    async def _stale_counter(**kwargs):
+        seen.append(kwargs)
+        # a stale-low counter read; the authoritative spend is what the budget must be judged on
+        return 0.0 if kwargs.get("max_budget") is None else kwargs["fallback_spend"]
+
+    monkeypatch.setattr(proxy_server, "get_current_spend", _stale_counter, raising=False)
+    token = _token(user_spend=1900.0, user_max_budget=50.0)
+
+    assert await is_token_within_budget_for_model(model="paid-model", valid_token=token, llm_router=_router()) is False
+    assert [call["max_budget"] for call in seen] == [50.0]
+
+
+@pytest.mark.asyncio
+async def test_check_fails_closed_when_the_spend_lookup_breaks(monkeypatch):
+    from litellm.proxy import proxy_server
+
+    async def _boom(**kwargs):
+        raise RuntimeError("spend counter unavailable")
+
+    monkeypatch.setattr(proxy_server, "get_current_spend", _boom, raising=False)
+    request = {"metadata": {"user_api_key_auth": _token(user_spend=1.0, user_max_budget=50.0)}}
+
+    assert await ENFORCED(model="paid-model", request_kwargs=request, llm_router=_router()) is False
+
+
+@pytest.mark.asyncio
+async def test_router_skips_the_paid_fallback_target_when_over_budget():
+    from litellm.router_utils.fallback_event_handlers import _is_fallback_target_within_budget
+
+    router = Router(
+        model_list=[FREE_MODEL, PAID_MODEL],
+        fallbacks=[{"free-model": ["paid-model"]}],
+        fallback_budget_check=ENFORCED,
+    )
+    over = {"metadata": {"user_api_key_auth": _token(user_spend=1900.0, user_max_budget=50.0)}}
+    under = {"metadata": {"user_api_key_auth": _token(user_spend=1.0, user_max_budget=50.0)}}
+
+    assert await _is_fallback_target_within_budget(router, "paid-model", "free-model", over) is False
+    assert await _is_fallback_target_within_budget(router, "paid-model", "free-model", under) is True
+
+
+@pytest.mark.asyncio
+async def test_router_without_a_budget_check_attempts_every_fallback():
+    from litellm.router_utils.fallback_event_handlers import _is_fallback_target_within_budget
+
+    router = _router()  # fallback_budget_check defaults to None
+    over = {"metadata": {"user_api_key_auth": _token(user_spend=1900.0, user_max_budget=50.0)}}
+
+    assert await _is_fallback_target_within_budget(router, "paid-model", "free-model", over) is True

tests/test_litellm/proxy/test_proxy_server.py

@@ -13387,6 +13387,44 @@ async def test_load_config_router_authorizes_fallback_targets_against_the_callin
     assert router.fallback_access_check is router_fallback_access_check
 
 
+@pytest.mark.asyncio
+async def test_load_config_router_budget_checks_fallback_targets_against_the_calling_key(tmp_path, monkeypatch):
+    """A config-loaded router refuses a paid fallback target for an over-budget caller."""
+    from litellm.proxy import proxy_server
+    from litellm.proxy._types import UserAPIKeyAuth
+    from litellm.proxy.proxy_server import ProxyConfig
+
+    config_file = tmp_path / "config.yaml"
+    config_file.write_text(
+        yaml.dump({"model_list": [{"model_name": "m", "litellm_params": {"model": "openai/m", "api_key": "k"}}]})
+    )
+
+    router, _, _ = await ProxyConfig().load_config(router=None, config_file_path=str(config_file))
+
+    over_budget = {
+        "metadata": {
+            "user_api_key_auth": UserAPIKeyAuth(
+                api_key="hashed", token="hashed", user_id="u1", user_spend=99.0, user_max_budget=1.0
+            )
+        }
+    }
+    under_budget = {
+        "metadata": {
+            "user_api_key_auth": UserAPIKeyAuth(
+                api_key="hashed", token="hashed", user_id="u1", user_spend=0.0, user_max_budget=100.0
+            )
+        }
+    }
+
+    # off by default: the paid fallback is still attempted for an over-budget caller
+    monkeypatch.setattr(proxy_server, "general_settings", {}, raising=False)
+    assert await router.fallback_budget_check(model="m", request_kwargs=over_budget, llm_router=router) is True
+
+    monkeypatch.setattr(proxy_server, "general_settings", {"enforce_fallback_budget": True}, raising=False)
+    assert await router.fallback_budget_check(model="m", request_kwargs=over_budget, llm_router=router) is False
+    assert await router.fallback_budget_check(model="m", request_kwargs=under_budget, llm_router=router) is True
+
+
 @pytest.mark.asyncio
 async def test_load_config_user_api_key_cache_max_size_keeps_more_than_200_entries(tmp_path, monkeypatch):
     """The auth cache used to be pinned at InMemoryCache's 200 entry default, so a

tests/test_litellm/router_utils/test_fallback_event_handlers.py

@@ -27,6 +27,7 @@ class StreamingWrapper:
 
 class FakeRouter:
     fallback_access_check = None
+    fallback_budget_check = None
 
     def log_retry(self, kwargs, e):
         return kwargs
@@ -37,6 +38,7 @@ class FakeRouter:
 
 class AlwaysFailRouter:
     fallback_access_check = None
+    fallback_budget_check = None
 
     def log_retry(self, kwargs, e):
         return kwargs
@@ -101,6 +103,7 @@ async def test_run_async_fallback_raises_when_all_fallbacks_fail():
 
 class RecordingRouter:
     fallback_access_check = None
+    fallback_budget_check = None
 
     def __init__(self):
         self.received_kwargs = None
@@ -162,6 +165,7 @@ async def test_run_async_fallback_skips_original_model_group():
 
 class AttemptRecordingRouter:
     fallback_access_check = None
+    fallback_budget_check = None
 
     def __init__(self):
         self.attempted_model_groups = []
@@ -471,6 +475,8 @@ class AccessCheckedRouter(AttemptRecordingRouter):
         self.allowed_models = allowed_models
         self.access_checks = []
 
+    fallback_budget_check = None
+
     async def fallback_access_check(self, *, model, request_kwargs, llm_router):
         self.access_checks.append((model, request_kwargs["metadata"]["user_api_key"], llm_router is self))
         return model in self.allowed_models
@@ -542,6 +548,7 @@ async def test_run_async_fallback_does_not_consult_access_check_for_same_model_g
 
 class RecordingFailRouter:
     fallback_access_check = None
+    fallback_budget_check = None
 
     def __init__(self):
         self.attempted_models = []
@@ -1053,6 +1060,7 @@ class TestTriggerCooldownForFailedDeployment:
 class TestRunAsyncFallbackTriggersCooldown:
     class RouterWithLoggingKwarg:
         fallback_access_check = None
+        fallback_budget_check = None
 
         def __init__(self):
             self.cooldown_time = 60.0