Merge 1c6e2f7959 into e6c4580a31

litellm/__init__.py

@@ -119,6 +119,8 @@ _custom_logger_compatible_callbacks_literal = Literal[
     "litellm_agent",
     "dynamic_rate_limiter",
     "dynamic_rate_limiter_v3",
+    "model_based_tag_rate_limits_hook",
+    "global_tag_rate_limits_hook",
     "langsmith",
     "prometheus",
     "otel",
@@ -391,6 +393,9 @@ cache: Optional["Cache"] = None  # cache object <- use this - https://docs.litel
 default_in_memory_ttl: Optional[float] = None
 default_redis_ttl: Optional[float] = None
 default_redis_batch_cache_expiry: Optional[float] = None
+model_based_tag_rate_limits_max_in_memory_cache_size: Optional[int] = None
+global_tag_rate_limits: Optional["TagRateLimits"] = None
+global_tag_rate_limits_max_in_memory_cache_size: Optional[int] = None
 model_alias_map: Dict[str, str] = {}
 model_group_settings: Optional["ModelGroupSettings"] = None
 max_budget: float = 0.0  # set the max budget across all providers

litellm/caching/in_memory_cache.py

@@ -163,7 +163,12 @@ class InMemoryCache(BaseCache):
             return
 
         self.cache_dict[key] = value
-        if self.allow_ttl_override(key):  # if ttl is not set, set it to default ttl
+        # refresh_ttl bypasses allow_ttl_override's "leave a still-live ttl
+        # alone" guard -- a caller only sets it for a counter whose ttl must
+        # keep extending on every write (e.g. a concurrency reservation's
+        # crash-safety-net ttl), never for one that must stay fixed to its
+        # original epoch window (e.g. a fixed-period rate-limit bucket).
+        if kwargs.get("refresh_ttl") or self.allow_ttl_override(key):  # if ttl is not set, set it to default ttl
             if "ttl" in kwargs and kwargs["ttl"] is not None:
                 self.ttl_dict[key] = time.time() + float(kwargs["ttl"])
                 heapq.heappush(self.expiration_heap, (self.ttl_dict[key], key))
@@ -248,7 +253,15 @@ class InMemoryCache(BaseCache):
     ) -> list[float] | None:
         results: Final = []
         for increment in increment_list:
-            result = await self.async_increment(increment["key"], increment["increment_value"], **kwargs)
+            # Each operation's own ttl must reach set_cache, or a key with no
+            # live ttl yet falls through to the cache's short default_ttl
+            # instead of the caller's real (often much longer) window --
+            # allow_ttl_override already leaves an existing, still-live ttl
+            # untouched on a later increment, so passing this through on
+            # every call is safe: it only ever takes effect the first time.
+            result = await self.async_increment(
+                increment["key"], increment["increment_value"], ttl=increment.get("ttl"), **kwargs
+            )
             results.append(result)
         return results
 

litellm/integrations/custom_logger.py

@@ -187,6 +187,16 @@ class CustomLogger:  # https://docs.litellm.ai/docs/observability/custom_callbac
     async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
         pass
 
+    async def async_release_disconnect_state_hook(self, request_data: Mapping[str, object]) -> None:
+        """
+        Called when a client disconnects mid-request, for a callback that reserved
+        per-request state outside async_log_success_event/async_log_failure_event
+        (e.g. a concurrency slot admitted before the first response chunk) -- those
+        two callbacks never run for a client disconnect, so a callback relying on
+        them alone to release such state would otherwise leak it until its own
+        safety-net TTL.
+        """
+
     async def async_log_audit_log_event(self, audit_log: "StandardAuditLogPayload"):
         """Called when an audit log is created. Override in subclasses to handle."""
 

litellm/litellm_core_utils/litellm_logging.py

@@ -4476,6 +4476,43 @@ def _init_custom_logger_compatible_class(
                 dynamic_rate_limiter_obj_v3.update_variables(llm_router=llm_router)
             _in_memory_loggers.append(dynamic_rate_limiter_obj_v3)
             return dynamic_rate_limiter_obj_v3
+        elif logging_integration == "model_based_tag_rate_limits_hook":
+            from litellm.proxy.hooks.model_based_tag_rate_limits_hook import (
+                _PROXY_ModelBasedTagRateLimitsHook,  # pyright: ignore[reportPrivateUsage]  # resolved by name like every other opt-in callback here
+            )
+
+            for callback in _in_memory_loggers:
+                if isinstance(callback, _PROXY_ModelBasedTagRateLimitsHook):
+                    return callback
+
+            if internal_usage_cache is None:
+                raise Exception(f"Internal Error: Cache cannot be empty - internal_usage_cache={internal_usage_cache}")
+
+            model_based_tag_rate_limits_hook_obj: Final = _PROXY_ModelBasedTagRateLimitsHook(
+                internal_usage_cache=internal_usage_cache
+            )
+
+            if llm_router is not None and isinstance(llm_router, litellm.Router):
+                model_based_tag_rate_limits_hook_obj.update_variables(llm_router=llm_router)
+            _in_memory_loggers.append(model_based_tag_rate_limits_hook_obj)
+            return model_based_tag_rate_limits_hook_obj
+        elif logging_integration == "global_tag_rate_limits_hook":
+            from litellm.proxy.hooks.global_tag_rate_limits_hook import (
+                _PROXY_GlobalTagRateLimitsHook,  # pyright: ignore[reportPrivateUsage]  # resolved by name like every other opt-in callback here
+            )
+
+            for callback in _in_memory_loggers:
+                if isinstance(callback, _PROXY_GlobalTagRateLimitsHook):
+                    return callback
+
+            if internal_usage_cache is None:
+                raise Exception(f"Internal Error: Cache cannot be empty - internal_usage_cache={internal_usage_cache}")
+
+            global_tag_rate_limits_hook_obj: Final = _PROXY_GlobalTagRateLimitsHook(
+                internal_usage_cache=internal_usage_cache
+            )
+            _in_memory_loggers.append(global_tag_rate_limits_hook_obj)
+            return global_tag_rate_limits_hook_obj
         elif logging_integration == "langtrace":
             if "LANGTRACE_API_KEY" not in os.environ:
                 raise ValueError("LANGTRACE_API_KEY not found in environment variables")
@@ -4916,6 +4953,24 @@ def get_custom_logger_compatible_class(
                 if isinstance(callback, _PROXY_DynamicRateLimitHandlerV3):
                     return callback
 
+        elif logging_integration == "model_based_tag_rate_limits_hook":
+            from litellm.proxy.hooks.model_based_tag_rate_limits_hook import (
+                _PROXY_ModelBasedTagRateLimitsHook,  # pyright: ignore[reportPrivateUsage]  # resolved by name like every other opt-in callback here
+            )
+
+            for callback in _in_memory_loggers:
+                if isinstance(callback, _PROXY_ModelBasedTagRateLimitsHook):
+                    return callback
+
+        elif logging_integration == "global_tag_rate_limits_hook":
+            from litellm.proxy.hooks.global_tag_rate_limits_hook import (
+                _PROXY_GlobalTagRateLimitsHook,  # pyright: ignore[reportPrivateUsage]  # resolved by name like every other opt-in callback here
+            )
+
+            for callback in _in_memory_loggers:
+                if isinstance(callback, _PROXY_GlobalTagRateLimitsHook):
+                    return callback
+
         elif logging_integration == "langtrace":
             from litellm.integrations.opentelemetry import OpenTelemetry
 

litellm/proxy/common_request_processing.py

@@ -33,6 +33,7 @@ from litellm.constants import (
     UNSAFE_PROXY_RESPONSE_HEADERS,
 )
 from litellm.integrations.custom_guardrail import CustomGuardrail
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.core_helpers import get_or_create_metadata_bucket, is_expected_client_error
 from litellm.litellm_core_utils.dd_tracing import NullTracer, tracer
 from litellm.litellm_core_utils.get_supported_openai_params import (
@@ -397,6 +398,32 @@ async def _bill_partial_streamed_spend_on_disconnect(request_data: dict, respons
     return True
 
 
+async def _release_disconnect_state_on_all_callbacks(request_data: Mapping[str, object]) -> None:
+    """
+    A client disconnect throws GeneratorExit/CancelledError into the streaming
+    generator, so neither the success nor failure logging callback runs for it
+    (see the callers of this function). A callback that reserves per-request
+    state outside of those two callbacks (e.g. a concurrency slot admitted
+    before the first chunk) would otherwise leak that state until its own
+    safety-net TTL. Give every registered callback a chance to release such
+    state via the optional, default-no-op ``async_release_disconnect_state_hook``.
+
+    Only ``CustomLogger`` instances are considered, never raw string entries:
+    by the time a request can reach this proxy-only cleanup path, startup's
+    ``ProxyLogging._init_litellm_callbacks`` has already replaced every string
+    entry in ``litellm.callbacks`` with its initialized instance in place.
+    """
+    for callback in litellm.callbacks:
+        if not isinstance(callback, CustomLogger):
+            continue
+        try:
+            await callback.async_release_disconnect_state_hook(request_data)
+        except Exception as e:  # noqa: BLE001  # one callback's cleanup must never block another's or the response teardown
+            verbose_proxy_logger.debug(
+                "Failed to run async_release_disconnect_state_hook for %s: %s", type(callback).__name__, e
+            )
+
+
 async def _cancel_pending_gather_tasks(tasks: list["asyncio.Task[Any]"]) -> None:
     pending_tasks: Final = [task for task in tasks if not task.done()]
     for task in pending_tasks:
@@ -1454,6 +1481,7 @@ async def _cancel_llm_call_on_client_disconnect(
 async def _await_llm_call_cancelling_on_disconnect(
     request: Request,
     llm_api_call: "asyncio.Future[_LlmCallT]",
+    request_data: Mapping[str, object],
 ) -> _LlmCallT:
     disconnect_event: Final = asyncio.Event()
     monitor: Final = asyncio.create_task(_cancel_llm_call_on_client_disconnect(request, llm_api_call, disconnect_event))
@@ -1461,6 +1489,14 @@ async def _await_llm_call_cancelling_on_disconnect(
         return await llm_api_call
     except asyncio.CancelledError:
         if disconnect_event.is_set():
+            # This cancellation never reaches litellm.utils.wrapper_async's own
+            # except block (asyncio.CancelledError is a BaseException, not an
+            # Exception, since Python 3.8), so async_log_failure_event never
+            # fires for it -- the same gap async_release_disconnect_state_hook
+            # was added for on the streaming path (see
+            # _finalize_streaming_generator_cleanup), just reached here via a
+            # cancelled non-streaming call instead of a mid-stream disconnect.
+            await _release_disconnect_state_on_all_callbacks(request_data)
             raise HTTPException(
                 status_code=499,
                 detail=_CLIENT_DISCONNECT_DETAIL,
@@ -1967,7 +2003,10 @@ class ProxyBaseLLMRequestProcessing:
             )
         except ProxyRateLimitError as original_exc:
             original_model: Final = self.data.get("model")
-            if not original_model or not llm_router or self.data.get("disable_fallbacks"):
+            cross_model_scope: Final = (
+                isinstance(original_exc.detail, Mapping) and original_exc.detail.get("cross_model_scope") is True
+            )
+            if not original_model or not llm_router or self.data.get("disable_fallbacks") or cross_model_scope:
                 raise
 
             fallback_models: Final = self._resolve_fallback_models(
@@ -2006,7 +2045,19 @@ class ProxyBaseLLMRequestProcessing:
                             route_type=route_type,
                             llm_router=llm_router,
                         )
-                    except ProxyRateLimitError:
+                    except ProxyRateLimitError as fallback_exc:
+                        # A fallback attempt's own rejection can carry the
+                        # identical cross_model_scope marker (this fallback
+                        # model is itself covered by the same apply_to_models
+                        # chain-wide cap) -- continuing to the next fallback
+                        # would silently serve the request through a model
+                        # outside that cap, defeating it just as much as not
+                        # checking the original exception would.
+                        if (
+                            isinstance(fallback_exc.detail, Mapping)
+                            and fallback_exc.detail.get("cross_model_scope") is True
+                        ):
+                            raise
                         continue
             except BaseException:
                 self.data["model"] = original_model
@@ -2285,7 +2336,9 @@ class ProxyBaseLLMRequestProcessing:
 
         try:
             if general_settings.get("cancel_on_disconnect", False):
-                responses = await _await_llm_call_cancelling_on_disconnect(request, llm_responses)
+                responses = await _await_llm_call_cancelling_on_disconnect(  # rebind-ok: assigned in exactly one of these two mutually exclusive branches
+                    request, llm_responses, self.data
+                )
             else:
                 responses = await llm_responses
         finally:
@@ -3364,6 +3417,8 @@ class ProxyBaseLLMRequestProcessing:
                     and user_api_key_dict is not None
                 ):
                     await proxy_logging_obj._arelease_max_parallel_requests_on_disconnect(user_api_key_dict)
+                if not success_event_owns_slot_release:
+                    await _release_disconnect_state_on_all_callbacks(request_data)
 
             if hasattr(response, "aclose"):
                 try:

litellm/proxy/hooks/global_tag_rate_limits_hook.py

@@ -0,0 +1,791 @@
+"""
+Tag-scoped token, request, dollar, and concurrency rate limits declared once,
+globally, in `litellm_settings.global_tag_rate_limits` and enforced in
+`async_pre_call_hook`, before Router does any routing -- so a limit applies
+regardless of which model or fallback chain the request ends up hitting.
+
+Model-independent sibling of `model_based_tag_rate_limits_hook`, which
+enforces the same `TagRateLimitEntry` shape per-deployment instead. Reuses
+that sibling's shared helpers (`entry_applies`, the atomic Lua scripts, cache
+partitioning, bucket-key hashing) from `tag_rate_limits_shared.py`, but has
+its own smaller admission/accounting engine with no routing-group or
+per-deployment concerns.
+
+Three entry-level knobs: `apply_to_key_alias` scopes an entry to specific
+virtual-key aliases; `apply_to_models` scopes it to specific caller-facing
+model names, letting one entry cap a whole fallback chain as a unit (a
+rejection then carries `detail["cross_model_scope"] = True` so
+`_pre_call_with_fallbacks` re-raises instead of silently admitting the
+request through an unlisted fallback model); `scope_by_key_hash` controls
+whether matching keys share one bucket or each gets its own.
+
+`data["litellm_logging_obj"]` does exist by the time `async_pre_call_hook`
+runs, but `_pre_call_with_fallbacks` re-runs the whole pre-call pipeline
+(building a fresh `Logging` object each time) once per fallback model on the
+same `litellm_call_id`, so unlike `model_based_tag_rate_limits_hook`'s
+per-Router-hop reservations, a stash keyed to one attempt's own
+`model_call_details` wouldn't survive to a later attempt. Per-request state
+instead lives on a `ContextVar`-based stash (the same pattern
+`parallel_request_limiter_v3.py` uses for the identical admission-to-release
+problem), keyed by `litellm_call_id` so a nested LiteLLM call sharing the
+same inherited context (an LLM-judge guardrail, for example) gets its own
+isolated entry instead of releasing the outer call's still-pending
+reservation early. Confirmed live: a real streaming client disconnect
+correctly releases its concurrency reservation through this mechanism.
+"""
+
+import asyncio
+from collections.abc import Callable, Mapping, Sequence
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from datetime import datetime
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final, TypeAlias
+
+import litellm
+from litellm._logging import verbose_proxy_logger
+from litellm.caching.dual_cache import DualCache
+from litellm.caching.in_memory_cache import InMemoryCache
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError
+from litellm.proxy.hooks.parallel_request_limiter_v3 import (
+    _PROXY_MaxParallelRequestsHandler_v3,  # pyright: ignore[reportPrivateUsage]  # shared private helper, reused by model_based_tag_rate_limits_hook too
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    ATOMIC_UNITS as _ATOMIC_UNITS,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    BACKGROUND_TASKS as _BACKGROUND_TASKS,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    CONCURRENCY_MIN_SAFETY_TTL_SECONDS as _CONCURRENCY_MIN_SAFETY_TTL_SECONDS,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    LIMIT_UNITS as _LIMIT_UNITS,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    TAG_RL_CHECK_AND_INCR_SCRIPT,
+    TAG_RL_DECR_FLOOR_ZERO_SCRIPT,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    UNIT_TO_GROUP_FIELD as _UNIT_TO_GROUP_FIELD,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    UNIT_TO_RATE_LIMIT_TYPE as _UNIT_TO_RATE_LIMIT_TYPE,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    LimitUnit as _LimitUnit,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    PartitionKey as _PartitionKey,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    PartitionOperations as _PartitionOperations,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    bucket_ttl_seconds as _bucket_ttl_seconds,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    entry_applies as _entry_applies,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    extract_identity as _extract_identity,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    extract_key_alias as _extract_key_alias,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    extract_key_hash as _extract_key_hash,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    fixed_length_identity as _fixed_length_identity,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    order_tags_for_identity_resolution as _order_tags_for_identity_resolution,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    partition_key as _partition_key,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    policy_fingerprint as _policy_fingerprint,
+)
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    resolve_authoritative_metadata_variable_name as _resolve_authoritative_metadata_variable_name,
+)
+from litellm.proxy.utils import InternalUsageCache
+from litellm.router_strategy.tag_based_routing import (
+    _get_tags_from_request_kwargs,  # pyright: ignore[reportPrivateUsage]  # shared private helper, reused by model_based_tag_rate_limits_hook too
+)
+from litellm.types.caching import RedisPipelineIncrementOperation
+from litellm.types.router import TagRateLimitEntry, TagRateLimits
+
+if TYPE_CHECKING:
+    from opentelemetry.trace import Span as _Span
+
+    Span: TypeAlias = _Span
+else:
+    Span: TypeAlias = object
+
+
+def _entry_applies_any_admitted_model(
+    entry: TagRateLimitEntry, tags: Sequence[str], key_alias: str | None, admitted_models: frozenset[str]
+) -> bool:
+    """Same as `_entry_applies`, except an `apply_to_models`-scoped entry
+    counts as applying if ANY model an admission attempt for this call_id
+    saw was in scope -- not just whichever model the call ultimately served.
+    A `_pre_call_with_fallbacks` retry re-admits with a different model for
+    the same call_id, and an entry that matched an earlier attempt must
+    still get its success-time accounting."""
+    if not admitted_models:
+        return _entry_applies(entry, tags, key_alias, None)
+    return any(_entry_applies(entry, tags, key_alias, model) for model in admitted_models)
+
+
+def _hash_tag(entry: TagRateLimitEntry, unit: _LimitUnit, tag_value: str, key_hash: str | None) -> str:
+    """Namespaced under `tag_rl:global:` so it never collides with
+    `model_based_tag_rate_limits_hook`'s own `tag_rl:{model_group}:...` keys."""
+    key_suffix: Final = f":key:{key_hash}" if key_hash is not None else ""
+    policy_suffix: Final = f":policy:{_policy_fingerprint(entry)}"
+    return f"tag_rl:global:{unit}:{entry.name}:{entry.tag_id}:{_fixed_length_identity(tag_value)}{key_suffix}{policy_suffix}"
+
+
+def _bucket_key(
+    entry: TagRateLimitEntry, unit: _LimitUnit, tag_value: str, bucket_id: int, key_hash: str | None
+) -> str:
+    return f"{{{_hash_tag(entry, unit, tag_value, key_hash)}}}:{bucket_id}"
+
+
+def _inflight_key(entry: TagRateLimitEntry, unit: _LimitUnit, tag_value: str, key_hash: str | None) -> str:
+    return f"{{{_hash_tag(entry, unit, tag_value, key_hash)}}}:inflight"
+
+
+@dataclass(frozen=True, slots=True)
+class _ClassifiedGlobalCheck:
+    unit: _LimitUnit
+    entry: TagRateLimitEntry
+    tag_value: str
+    key: str
+    is_atomic: bool
+
+
+@dataclass(frozen=True, slots=True)
+class _CachePartition:
+    internal_usage_cache: InternalUsageCache
+    v3: _PROXY_MaxParallelRequestsHandler_v3
+
+
+@dataclass(slots=True)
+class _GlobalTagRateLimitStash:
+    """Per-call bookkeeping `async_pre_call_hook` hands to that same call's
+    success/failure/disconnect callbacks -- see module docstring for why this
+    lives on a `ContextVar`, not `model_call_details`.
+
+    Keyed by `litellm_call_id` rather than one shared mutable instance so a
+    nested LiteLLM call (an LLM-judge guardrail) that mints its own call id
+    but inherits the same context doesn't release the outer call's
+    still-pending reservation early.
+    """
+
+    admission_time: float | None = None
+    # Every model any admission attempt for this call_id has classified
+    # entries against, accumulated rather than overwritten: a
+    # _pre_call_with_fallbacks retry re-runs admission with a *different*
+    # model for the same call_id, and an apply_to_models entry that matched
+    # an earlier attempt must still get its accounting at success time even
+    # though the request ultimately serves from a later attempt's model.
+    admitted_models: frozenset[str] = field(default_factory=frozenset)
+    pending_concurrency_keys: list[tuple[str, _PartitionKey]] = field(default_factory=list)  # mutable-ok: queue
+    # Keys already charged for this call_id, so a fallback retry (same
+    # litellm_call_id, different model) renews instead of double-charging.
+    charged_request_keys: list[str] = field(default_factory=list)  # mutable-ok: see comment above
+    # key_hash of whoever first claimed this stash. litellm_call_id is
+    # caller-controlled (x-litellm-call-id), so only a later admission with
+    # the same authenticated key_hash may renew this stash's charges.
+    owner_key_hash: str | None = None
+
+
+# Sentinel key for a call with no litellm_call_id at all (claim and lookup
+# both fall back to this same key, so behavior for that degenerate case is
+# unchanged: everything without a call id still shares one bucket).
+_NO_CALL_ID: Final = "<no-call-id>"
+
+_StashByCallId: TypeAlias = dict[
+    str, _GlobalTagRateLimitStash
+]  # mutable-ok: per-call-id entries added over a request's lifetime, see class docstring
+
+_request_stash: Final[ContextVar[_StashByCallId | None]] = ContextVar(
+    "global_tag_rate_limits_request_stash", default=None
+)
+
+
+def _claim_stash_for_data(data: Mapping[str, object]) -> _GlobalTagRateLimitStash:
+    by_call_id: _StashByCallId | None = _request_stash.get()  # rebind-ok: lazily initialized below if never set
+    if by_call_id is None:
+        by_call_id = {}  # rebind-ok: see above  # mutable-ok: see _StashByCallId
+        _request_stash.set(by_call_id)
+    owner_call_id: Final = data.get("litellm_call_id")
+    key: Final = owner_call_id if isinstance(owner_call_id, str) else _NO_CALL_ID
+    stash = by_call_id.get(key)  # rebind-ok: reassigned just below when newly created
+    if stash is None:
+        stash = _GlobalTagRateLimitStash()  # rebind-ok: see above
+        by_call_id[key] = stash  # mutable-ok: see class docstring
+    return stash
+
+
+def _stash_for_call(litellm_call_id: str | None) -> _GlobalTagRateLimitStash | None:
+    by_call_id: Final = _request_stash.get()
+    if by_call_id is None:
+        return None
+    key: Final = litellm_call_id if litellm_call_id is not None else _NO_CALL_ID
+    return by_call_id.get(key)
+
+
+def _call_id_from_kwargs(kwargs: Mapping[str, object]) -> str | None:
+    call_id: Final = kwargs.get("litellm_call_id")
+    return call_id if isinstance(call_id, str) else None
+
+
+def _resolve_max_in_memory_cache_size() -> int | None:
+    """Same shape as `model_based_tag_rate_limits_hook`'s own function, reading
+    this hook's own `litellm_settings` knob instead."""
+    configured: Final = litellm.global_tag_rate_limits_max_in_memory_cache_size
+    if isinstance(configured, int) and not isinstance(configured, bool) and configured > 0:
+        return configured
+    if configured is not None:
+        verbose_proxy_logger.warning(
+            "global_tag_rate_limits_hook: global_tag_rate_limits_max_in_memory_cache_size=%r is not a positive "
+            "integer; falling back to the default in-memory cache size.",
+            configured,
+        )
+    return None
+
+
+class _PROXY_GlobalTagRateLimitsHook(  # pyright: ignore[reportUnusedClass]  # only referenced via the deferred import in litellm_logging.py's callback resolver; basedpyright doesn't trace that usage
+    CustomLogger
+):
+    def __init__(
+        self,
+        internal_usage_cache: DualCache,
+        time_provider: Callable[[], datetime] | None = None,
+    ) -> None:
+        self._redis_cache: Final = internal_usage_cache.redis_cache
+        self._time_provider = time_provider or datetime.now
+        self._partitions: dict[_PartitionKey, _CachePartition] = {}  # mutable-ok: lazily memoized; see _partition_for
+        self._partitions_lock = asyncio.Lock()
+        default_partition: Final = self._build_partition(_resolve_max_in_memory_cache_size())
+        self._partitions[None] = default_partition
+        self.internal_usage_cache = default_partition.internal_usage_cache
+        self._lock = asyncio.Lock()
+        redis_cache: Final = self._redis_cache
+        self._check_and_incr_script = (
+            redis_cache.async_register_script(TAG_RL_CHECK_AND_INCR_SCRIPT) if redis_cache is not None else None
+        )
+        self._decr_floor_zero_script = (
+            redis_cache.async_register_script(TAG_RL_DECR_FLOOR_ZERO_SCRIPT) if redis_cache is not None else None
+        )
+        self._config_cache_key: object | None = None
+        self._config: TagRateLimits | None = None
+
+    def _refresh_config(self) -> TagRateLimits | None:
+        """Re-validates `litellm.global_tag_rate_limits` whenever the object
+        identity changes (a config reload replaces it wholesale via
+        `setattr(litellm, key, value)`), so a hot-reloaded config takes effect
+        on the very next request with no staleness window and no TTL to tune."""
+        raw: Final = getattr(litellm, "global_tag_rate_limits", None)
+        if raw is not self._config_cache_key:
+            self._config = TagRateLimits.model_validate(raw) if raw else None
+            self._config_cache_key = raw
+        return self._config
+
+    def _build_partition(self, cache_size_override: int | None) -> _CachePartition:
+        dual_cache: Final = DualCache(
+            in_memory_cache=InMemoryCache(max_size_in_memory=cache_size_override),
+            redis_cache=self._redis_cache,
+        )
+        cache: Final = InternalUsageCache(dual_cache=dual_cache)
+        return _CachePartition(
+            internal_usage_cache=cache,
+            v3=_PROXY_MaxParallelRequestsHandler_v3(cache, time_provider=self._time_provider),
+        )
+
+    async def _partition_for(self, partition_key: _PartitionKey) -> _CachePartition:
+        existing: Final = self._partitions.get(partition_key)
+        if existing is not None:
+            return existing
+        async with self._partitions_lock:
+            existing_after_lock: Final = self._partitions.get(partition_key)
+            if existing_after_lock is not None:
+                return existing_after_lock
+            cache_size_override: Final = partition_key[-1] if partition_key is not None else None
+            built: Final = self._build_partition(cache_size_override)
+            self._partitions[partition_key] = (
+                built  # mutable-ok: lazily memoized per distinct partition key, guarded by _partitions_lock above
+            )
+            return built
+
+    async def _check_and_increment_one(
+        self, cache: InternalUsageCache, key: str, limit: float, increment: float, ttl: int, refresh_ttl: bool
+    ) -> tuple[bool, float]:
+        if self._check_and_incr_script is not None:
+            raw: Final = await self._check_and_incr_script(
+                keys=(key,), args=(limit, increment, ttl, 1 if refresh_ttl else 0)
+            )
+            return bool(raw[0]), float(raw[1])
+        async with self._lock:
+            current_value: Final = await cache.async_get_cache(key=key, litellm_parent_otel_span=None)
+            current: Final = float(current_value) if current_value is not None else 0.0
+            if current + increment > limit:
+                return False, current
+            new_value: Final = current + increment
+            await cache.async_set_cache(
+                key=key, value=new_value, ttl=ttl, refresh_ttl=refresh_ttl, litellm_parent_otel_span=None
+            )
+            return True, new_value
+
+    async def _decrement_floor_zero(self, cache: InternalUsageCache, key: str, delta: float) -> None:
+        if self._decr_floor_zero_script is not None:
+            await self._decr_floor_zero_script(keys=(key,), args=(delta,))
+            return
+        async with self._lock:
+            current_value: Final = await cache.async_get_cache(key=key, litellm_parent_otel_span=None)
+            current: Final = float(current_value) if current_value is not None else 0.0
+            await cache.async_set_cache(key=key, value=max(0.0, current + delta), litellm_parent_otel_span=None)
+
+    async def _atomic_check_and_increment(
+        self,
+        checks: Sequence[tuple[InternalUsageCache, str, float, float, int, bool]],
+    ) -> tuple[int | None, tuple[float, ...]]:
+        """All-or-nothing atomic admission across `checks`: on a rejection,
+        refunds every check admitted earlier in this batch."""
+        if not checks:
+            return None, ()
+        admitted_values: Final = []  # mutable-ok: sequential async accumulator, discardable on early rejection
+        for index, (cache, key, limit, increment, ttl, refresh_ttl) in enumerate(checks):
+            admitted = False
+            try:
+                admitted, value = await self._check_and_increment_one(cache, key, limit, increment, ttl, refresh_ttl)
+            finally:
+                if not admitted:
+                    await self._refund_admitted(checks, up_to_index=index)
+            if admitted:
+                admitted_values.append(value)  # mutable-ok: see accumulator comment above
+                continue
+            return index, (value,)
+        return None, tuple(admitted_values)
+
+    async def _refund_admitted(
+        self, checks: Sequence[tuple[InternalUsageCache, str, float, float, int, bool]], up_to_index: int
+    ) -> None:
+        for refund_index in range(up_to_index):
+            refund_cache, refund_key, _limit, refund_increment, _ttl, _refresh_ttl = checks[refund_index]
+            try:
+                await self._decrement_floor_zero(refund_cache, refund_key, -refund_increment)
+            except Exception as e:  # noqa: BLE001 - one failed refund must not block refunding the rest
+                verbose_proxy_logger.warning(
+                    "global_tag_rate_limits_hook: failed to refund %s on rollback: %s", refund_key, e
+                )
+
+    async def _release_keys(self, reservations: Sequence[tuple[str, _PartitionKey]]) -> None:
+        for key, partition_key in reservations:
+            try:
+                partition = await self._partition_for(partition_key)  # not Final: rebound each loop iteration
+                await self._decrement_floor_zero(partition.internal_usage_cache, key, -1.0)
+            except Exception as e:  # noqa: BLE001 - releasing a slot must never raise into the caller's request path
+                verbose_proxy_logger.warning(
+                    "global_tag_rate_limits_hook: failed to release concurrency slot %s: %s", key, e
+                )
+
+    @staticmethod
+    def _record_admitted_model(stash: _GlobalTagRateLimitStash, model: str | None, renewal_allowed: bool) -> None:
+        """Only called once this admission attempt has cleared every check
+        without raising -- a rejected attempt's model must never join
+        admitted_models, or a later successful attempt's accounting could
+        wrongly credit an apply_to_models entry that never actually admitted
+        this request under that model."""
+        if renewal_allowed and model is not None:
+            stash.admitted_models = stash.admitted_models | frozenset((model,))
+
+    @staticmethod
+    def _ttl_for(unit: _LimitUnit, entry: TagRateLimitEntry) -> int:
+        if unit == "concurrency":
+            requested_ttl: Final = entry.key_ttl_seconds if entry.key_ttl_seconds is not None else entry.period_seconds
+            return max(requested_ttl, _CONCURRENCY_MIN_SAFETY_TTL_SECONDS)
+        return _bucket_ttl_seconds(entry)
+
+    def _classify(
+        self,
+        config: TagRateLimits,
+        tags: Sequence[str],
+        key_alias: str | None,
+        key_hash: str | None,
+        now: float,
+        model: str | None,
+    ) -> tuple[_ClassifiedGlobalCheck, ...]:
+        classified: Final = []  # mutable-ok: sequential accumulator, immediately frozen into a tuple below
+        for unit in _LIMIT_UNITS:
+            group = getattr(config, _UNIT_TO_GROUP_FIELD[unit])
+            if group is None:
+                continue
+            for entry in group.limits:
+                tag_value = _extract_identity(tags, entry.tag_id)
+                if tag_value is None:
+                    continue
+                if not _entry_applies(entry, tags, key_alias, model):
+                    continue
+                effective_key_hash = key_hash if entry.scope_by_key_hash else None
+                if unit == "concurrency":
+                    key = _inflight_key(entry, unit, tag_value, key_hash=effective_key_hash)
+                    classified.append(
+                        _ClassifiedGlobalCheck(unit, entry, tag_value, key, is_atomic=True)
+                    )  # mutable-ok: see comment above
+                    continue
+                bucket_id = int(now) // entry.period_seconds
+                key = _bucket_key(entry, unit, tag_value, bucket_id, key_hash=effective_key_hash)
+                classified.append(  # mutable-ok: see comment above
+                    _ClassifiedGlobalCheck(unit, entry, tag_value, key, is_atomic=unit in _ATOMIC_UNITS)
+                )
+        return tuple(classified)
+
+    async def _read_only_values(
+        self, read_only_checks: Sequence[_ClassifiedGlobalCheck], parent_otel_span: Span | None
+    ) -> tuple[float | None, ...]:
+        if not read_only_checks:
+            return ()
+        indices_by_partition: Final[dict[_PartitionKey, list[int]]] = {}  # mutable-ok: grouped, reassembled below
+        for index, check in enumerate(read_only_checks):
+            partition_key = _partition_key(check.entry)
+            indices = indices_by_partition.setdefault(partition_key, [])  # mutable-ok: see above
+            indices.append(index)  # mutable-ok: see comment above
+        values_by_index: Final[dict[int, float | None]] = {}  # mutable-ok: see comment above
+        for partition_key, indices in indices_by_partition.items():
+            partition = await self._partition_for(partition_key)  # not Final: rebound each loop iteration
+            keys = [read_only_checks[i].key for i in indices]  # mutable-ok: async_batch_get_cache needs a real list
+            redis_cache = partition.internal_usage_cache.dual_cache.redis_cache
+            if redis_cache is not None:
+                redis_values: Mapping[str, object] = await redis_cache.async_batch_get_cache(
+                    key_list=keys, parent_otel_span=parent_otel_span
+                )
+                resolved = [redis_values.get(key) for key in keys]  # mutable-ok: needs a real list
+            else:
+                current_values = await partition.internal_usage_cache.async_batch_get_cache(
+                    keys=keys, parent_otel_span=parent_otel_span, local_only=True
+                )
+                missing = [None] * len(keys)  # mutable-ok: async_batch_get_cache requires a real list; see above
+                resolved = current_values if current_values is not None else missing
+            for i, value in zip(indices, resolved):
+                values_by_index[i] = value  # mutable-ok: see comment above
+        return tuple(values_by_index[i] for i in range(len(read_only_checks)))
+
+    def _raise_if_over_limit(
+        self,
+        read_only_checks: Sequence[_ClassifiedGlobalCheck],
+        current_values: Sequence[float | None],
+        model: str | None,
+    ) -> None:
+        for check, current_value in zip(read_only_checks, current_values):
+            current = float(current_value) if current_value is not None else 0.0
+            if current < check.entry.limit:
+                continue
+            self._raise_over_limit(check.unit, check.entry, check.tag_value, model, current=current)
+
+    def _raise_over_limit(
+        self, unit: _LimitUnit, entry: TagRateLimitEntry, tag_value: str, model: str | None, current: float
+    ) -> None:
+        verbose_proxy_logger.debug(
+            "global_tag_rate_limits_hook: OVER_LIMIT unit=%s name=%s tag_id=%s tag_value=%s current=%s limit=%s",
+            unit,
+            entry.name,
+            entry.tag_id,
+            tag_value,
+            current,
+            entry.limit,
+        )
+        raise ProxyRateLimitError(
+            detail={  # mutable-ok: async_log_failure_event and generic proxy exception rendering branch on isinstance(exc.detail, dict)
+                "error": "tag_rate_limit_exceeded",
+                "type": unit,
+                "tag_id": entry.tag_id,
+                # tag_value deliberately excluded: it can resolve from
+                # inherited_tags (server-assigned key/team/project metadata),
+                # and echoing it back would disclose that identity to the
+                # caller. verbose_proxy_logger.debug above still logs it
+                # server-side for observability.
+                "limit_name": entry.name,
+                "limit": entry.limit,
+                "period_seconds": entry.period_seconds,
+                # ProxyBaseLLMRequestProcessing._pre_call_with_fallbacks reads this:
+                # an apply_to_models entry caps an entire named chain as one unit, so
+                # retrying against a fallback model outside that list would silently
+                # defeat the very policy that just rejected this request.
+                **({"cross_model_scope": True} if entry.apply_to_models is not None else {}),  # mutable-ok: see above
+            },
+            headers={"retry-after": str(entry.period_seconds)},  # mutable-ok: same as detail
+            rate_limit_type=_UNIT_TO_RATE_LIMIT_TYPE[unit],
+            model=model,
+            llm_provider="litellm_proxy",
+        )
+
+    async def async_pre_call_hook(
+        self,
+        user_api_key_dict: UserAPIKeyAuth,
+        cache: DualCache,
+        data: dict,  # mutable-ok: must match CustomLogger.async_pre_call_hook's own base signature exactly
+        call_type: str,
+    ) -> dict:  # mutable-ok: must match CustomLogger.async_pre_call_hook's own base signature exactly
+        config: Final = self._refresh_config()
+        if config is None:
+            return data
+
+        # _pre_call_with_fallbacks can re-run this pipeline once per fallback
+        # model on any ProxyRateLimitError, reusing the same litellm_call_id --
+        # see charged_request_keys for how a repeat run renews instead of
+        # re-charging.
+        stash: Final = _claim_stash_for_data(data)
+
+        # Not get_metadata_variable_name_from_kwargs (naive key-presence
+        # check): a caller can forge an empty (or None) litellm_metadata on
+        # an ordinary request to make that check pick it over the real,
+        # populated metadata the proxy wrote identity/tags into, seeing no
+        # tags at all and admitting past every configured limit.
+        metadata_variable_name: Final = _resolve_authoritative_metadata_variable_name(data)
+        tags: Final = _order_tags_for_identity_resolution(
+            _get_tags_from_request_kwargs(data, metadata_variable_name=metadata_variable_name),
+            data,
+            metadata_variable_name,
+        )
+        key_alias: Final = user_api_key_dict.key_alias
+        key_hash: Final = user_api_key_dict.api_key
+        model: Final = data.get("model") if isinstance(data.get("model"), str) else None
+
+        # First admission for this stash claims ownership; only a later one
+        # with the same key_hash may renew its charges (see owner_key_hash).
+        if stash.owner_key_hash is None:
+            stash.owner_key_hash = key_hash
+        renewal_allowed: Final = stash.owner_key_hash == key_hash
+
+        now: Final = self._time_provider().timestamp()
+        stash.admission_time = now
+        classified: Final = self._classify(config, tags, key_alias, key_hash, now, model)
+        if not classified:
+            self._record_admitted_model(stash, model, renewal_allowed)
+            return data
+
+        read_only_checks: Final = tuple(c for c in classified if not c.is_atomic)
+        atomic_checks: Final = tuple(c for c in classified if c.is_atomic)
+
+        current_values: Final = await self._read_only_values(read_only_checks, parent_otel_span=None)
+        self._raise_if_over_limit(read_only_checks, current_values, model)
+
+        if atomic_checks:
+            atomic_partitions_list: Final = []  # mutable-ok: sequential async lookups, one per atomic_checks entry
+            for check in atomic_checks:
+                atomic_partitions_list.append(
+                    await self._partition_for(_partition_key(check.entry))
+                )  # mutable-ok: see comment above
+            atomic_partitions: Final = tuple(atomic_partitions_list)
+            already_reserved_concurrency_keys: Final = frozenset(
+                key for key, _partition_key in stash.pending_concurrency_keys
+            )
+            failing_index, values = await self._atomic_check_and_increment(
+                tuple(
+                    (
+                        partition.internal_usage_cache,
+                        check.key,
+                        check.entry.limit,
+                        # A key already charged/reserved for this call_id (an
+                        # earlier fallback attempt for the same request) renews
+                        # at zero net cost instead of charging a second unit.
+                        0.0
+                        if renewal_allowed
+                        and (
+                            (check.unit == "requests" and check.key in stash.charged_request_keys)
+                            or (check.unit == "concurrency" and check.key in already_reserved_concurrency_keys)
+                        )
+                        else 1.0,
+                        self._ttl_for(check.unit, check.entry),
+                        check.unit == "concurrency",
+                    )
+                    for partition, check in zip(atomic_partitions, atomic_checks)
+                )
+            )
+            if failing_index is not None:
+                failing_check: Final = atomic_checks[failing_index]
+                self._raise_over_limit(
+                    failing_check.unit, failing_check.entry, failing_check.tag_value, model, current=values[0]
+                )
+
+            # Exclude already_reserved_concurrency_keys: that key renewed at
+            # zero cost above, so re-adding it would make release decrement
+            # twice for a counter only ever incremented once.
+            concurrency_reservations: Final = tuple(
+                (check.key, _partition_key(check.entry))
+                for check in atomic_checks
+                if check.unit == "concurrency" and check.key not in already_reserved_concurrency_keys
+            )
+            if concurrency_reservations:
+                stash.pending_concurrency_keys.extend(concurrency_reservations)  # mutable-ok: see field's own docstring
+
+            # Only recorded when renewal_allowed, so a call_id collision from
+            # a different key_hash can't contaminate the rightful owner's
+            # renewal tracking.
+            request_keys: Final = (
+                tuple(
+                    check.key
+                    for check in atomic_checks
+                    if check.unit == "requests" and check.key not in stash.charged_request_keys
+                )
+                if renewal_allowed
+                else ()
+            )
+            if request_keys:
+                stash.charged_request_keys.extend(request_keys)  # mutable-ok: see field's own docstring
+
+        self._record_admitted_model(stash, model, renewal_allowed)
+        return data
+
+    async def _release_pending_for_call_id(self, request_kwargs: Mapping[str, object]) -> None:
+        stash: Final = _stash_for_call(_call_id_from_kwargs(request_kwargs))
+        if stash is None or not stash.pending_concurrency_keys:
+            return
+        release_keys: Final = tuple(stash.pending_concurrency_keys)
+        stash.pending_concurrency_keys.clear()
+        await self._release_keys(release_keys)
+
+    async def async_release_disconnect_state_hook(self, request_data: Mapping[str, object]) -> None:
+        await self._release_pending_for_call_id(request_data)
+
+    async def async_post_call_failure_hook(
+        self,
+        request_data: dict,  # mutable-ok: must match CustomLogger.async_post_call_failure_hook's own base signature exactly
+        original_exception: Exception,
+        user_api_key_dict: UserAPIKeyAuth,
+        traceback_str: str | None = None,
+    ) -> None:
+        """
+        A request that never reaches Router (every fallback model also
+        rejected, or none configured) never runs the actual LLM call, so
+        neither async_log_success_event nor async_log_failure_event -- both
+        tied to that call's own wrapper -- ever fires for it. This is the
+        only remaining release path for a reservation from an earlier,
+        successful admission attempt in the same _pre_call_with_fallbacks
+        chain. litellm_call_id survives proxy/utils.py's own stripping here
+        (only litellm_logging_obj is popped), so the same ContextVar-based
+        stash lookup as the other release hooks still works.
+        """
+        await self._release_pending_for_call_id(request_data)
+
+    async def async_log_failure_event(
+        self,
+        kwargs: Mapping[str, object],
+        response_obj: object,
+        start_time: datetime | None,
+        end_time: datetime | None,
+    ) -> None:
+        # Always release regardless of which hook raised: this hook's own
+        # rejection never reserves a slot, so pending_concurrency_keys is
+        # already empty in that case and the check below no-ops; a rejection
+        # from model_based_tag_rate_limits_hook (same error marker) can still
+        # land after this hook already reserved its own slot.
+        await self._release_pending_for_call_id(kwargs)
+
+    async def async_log_success_event(
+        self,
+        kwargs: Mapping[str, object],
+        response_obj: object,
+        start_time: datetime | None,
+        end_time: datetime | None,
+    ) -> None:
+        stash: Final = _stash_for_call(_call_id_from_kwargs(kwargs))
+        if stash is not None and stash.pending_concurrency_keys:
+            release_keys: Final = tuple(stash.pending_concurrency_keys)
+            stash.pending_concurrency_keys.clear()
+            release_task: Final = asyncio.create_task(self._release_keys(release_keys))
+            _BACKGROUND_TASKS.add(release_task)  # mutable-ok: see _BACKGROUND_TASKS's own docstring
+            release_task.add_done_callback(_BACKGROUND_TASKS.discard)
+
+        config: Final = self._refresh_config()
+        if config is None:
+            return
+
+        standard_logging_object: Final = kwargs.get("standard_logging_object")
+        if not isinstance(standard_logging_object, dict):
+            return
+
+        # kwargs here is Logging.model_call_details, not the router's flat
+        # request kwargs admission sees: metadata/litellm_metadata are never
+        # top-level here, only nested under kwargs["litellm_params"] (see
+        # Logging.update_environment_variables).
+        litellm_params_raw: Final = kwargs.get("litellm_params")
+        litellm_params_for_metadata: Final[Mapping[str, object]] = (
+            litellm_params_raw if isinstance(litellm_params_raw, Mapping) else kwargs
+        )
+        metadata_variable_name: Final = _resolve_authoritative_metadata_variable_name(litellm_params_for_metadata)
+        key_hash: Final = _extract_key_hash(litellm_params_for_metadata, metadata_variable_name)
+        key_alias: Final = _extract_key_alias(litellm_params_for_metadata, metadata_variable_name)
+
+        tags: Final = _order_tags_for_identity_resolution(
+            _get_tags_from_request_kwargs(litellm_params_for_metadata, metadata_variable_name=metadata_variable_name),
+            litellm_params_for_metadata,
+            metadata_variable_name,
+        )
+        if not tags:
+            return
+
+        now: Final = (
+            stash.admission_time
+            if stash is not None and stash.admission_time is not None
+            else self._time_provider().timestamp()
+        )
+        admitted_models: Final = stash.admitted_models if stash is not None else frozenset()
+        increment_by_unit: Final[Mapping[_LimitUnit, float]] = MappingProxyType(
+            {
+                "tokens": float(standard_logging_object.get("total_tokens") or 0),
+                "dollars": float(standard_logging_object.get("response_cost") or 0),
+            }
+        )
+
+        operation_by_entry: Final = []  # mutable-ok: sequential accumulator over config groups, immediately used below
+        for unit in ("tokens", "dollars"):
+            group = getattr(config, _UNIT_TO_GROUP_FIELD[unit])
+            if group is None:
+                continue
+            for entry in group.limits:
+                tag_value = _extract_identity(tags, entry.tag_id)
+                if tag_value is None:
+                    continue
+                if not _entry_applies_any_admitted_model(entry, tags, key_alias, admitted_models):
+                    continue
+                increment_value = increment_by_unit[unit]
+                if increment_value == 0:
+                    continue
+                bucket_id = int(now) // entry.period_seconds
+                key_hash_for_entry = key_hash if entry.scope_by_key_hash else None
+                key = _bucket_key(entry, unit, tag_value, bucket_id, key_hash=key_hash_for_entry)
+                operation_by_entry.append(  # mutable-ok: see comment above
+                    (
+                        entry,
+                        RedisPipelineIncrementOperation(
+                            key=key, increment_value=increment_value, ttl=_bucket_ttl_seconds(entry)
+                        ),
+                    )
+                )
+
+        if not operation_by_entry:
+            return
+
+        operations_by_partition: Final[_PartitionOperations] = {}  # mutable-ok: grouped by cache partition below
+        for entry, operation in operation_by_entry:
+            partition_key = _partition_key(entry)
+            operations = operations_by_partition.setdefault(partition_key, [])  # mutable-ok: see above
+            operations.append(operation)  # mutable-ok: see comment above
+
+        for partition_key, group_operations in operations_by_partition.items():
+            partition = await self._partition_for(partition_key)  # not Final: rebound each loop iteration
+            accounting_task = asyncio.create_task(  # not Final: rebound each loop iteration
+                partition.v3.async_increment_tokens_with_ttl_preservation(
+                    pipeline_operations=group_operations, parent_otel_span=None
+                )
+            )
+            _BACKGROUND_TASKS.add(accounting_task)  # mutable-ok: see _BACKGROUND_TASKS's own docstring
+            accounting_task.add_done_callback(_BACKGROUND_TASKS.discard)

litellm/proxy/hooks/tag_rate_limits_shared.py

@@ -0,0 +1,401 @@
+"""
+Primitives shared by both tag-scoped rate-limit hooks:
+`model_based_tag_rate_limits_hook.py` (per-deployment limits nested under
+`model_info.tag_rate_limits`, admitted once per routing hop) and
+`global_tag_rate_limits_hook.py` (model-independent limits declared once
+under `litellm_settings.global_tag_rate_limits`, admitted once per request).
+
+Both hooks enforce the identical `TagRateLimitEntry` shape and need the same
+identity/scope extraction, policy fingerprinting, bucket-key hashing, and
+cache-partitioning primitives, so those live here rather than in either
+hook's own module -- keeping neither hook reaching into the other's private
+internals to reuse them. Every name here is this module's own public
+interface (no leading underscore): each hook imports what it needs aliased
+back to its own historical, underscore-prefixed local name (e.g.
+`entry_applies as _entry_applies`), so this is a genuine export, not a
+private symbol either hook reaches across a module boundary to grab.
+"""
+
+import asyncio
+import hashlib
+from collections.abc import Mapping, Sequence
+from types import MappingProxyType
+from typing import Final, Literal, TypeAlias
+
+from litellm.exceptions import RateLimitType
+from litellm.types.caching import RedisPipelineIncrementOperation
+from litellm.types.router import TagRateLimitEntry, TagRateLimitScope
+
+LimitUnit: TypeAlias = Literal["tokens", "requests", "dollars", "concurrency"]
+LIMIT_UNITS: Final[tuple[LimitUnit, ...]] = ("tokens", "requests", "dollars", "concurrency")
+
+# Units whose admission must be atomic (check-and-increment in one Redis
+# round trip) because the increment amount is known upfront (always 1).
+# tokens/dollars can't be: real usage is only known after the response, so
+# they stay a read-then-account-on-success check with a documented,
+# unavoidable admit-vs-account race.
+ATOMIC_UNITS: Final[frozenset[LimitUnit]] = frozenset({"requests", "concurrency"})
+
+UNIT_TO_GROUP_FIELD: Final[Mapping[LimitUnit, str]] = MappingProxyType(
+    {
+        "tokens": "token_limits",
+        "requests": "request_limits",
+        "dollars": "dollar_limits",
+        "concurrency": "concurrency_limits",
+    }
+)
+UNIT_TO_RATE_LIMIT_TYPE: Final[Mapping[LimitUnit, RateLimitType]] = MappingProxyType(
+    {
+        "tokens": RateLimitType.TOKENS,
+        "requests": RateLimitType.REQUESTS,
+        "dollars": RateLimitType.BUDGET,
+        "concurrency": RateLimitType.CONCURRENT_REQUESTS,
+    }
+)
+
+# Shared read-only fallback for an absent/None mapping (request_kwargs,
+# metadata, model_info, ...): avoids constructing a fresh mutable `{}` at
+# every one of these call sites just to immediately call `.get()` on it.
+EMPTY_MAPPING: Final[Mapping[str, object]] = MappingProxyType({})
+
+# `asyncio.create_task`'s own docs: "Save a reference to the result of this
+# function, to avoid a task disappearing mid-execution. The event loop only
+# keeps weak references to tasks. A task that isn't referenced elsewhere may
+# get garbage collected at any time, even before it's done." The success path
+# deliberately fires-and-forgets its concurrency release and its token/dollar
+# accounting increment (unlike the failure/disconnect paths, which await
+# concurrency release directly) to keep the hot success-response path from
+# waiting on a Redis round trip; by the time either background task would
+# run, the state it needs (popped pending keys, or the request's own usage
+# figures) is only available in that task's own closure, so a collected
+# task's work is unrecoverable, not just delayed. Holding a strong reference
+# here until each task's own completion callback discards it is the standard
+# fix, shared by every fire-and-forget task either hook creates.
+BACKGROUND_TASKS: Final[set["asyncio.Task[None]"]] = set()  # mutable-ok: see comment above
+
+# Floor for a concurrency reservation's self-heal TTL, regardless of the
+# configured period_seconds. A reservation that expires while its request is
+# still genuinely in flight silently admits requests past the limit; this
+# generous floor keeps that window far larger than any realistic request
+# duration, at the cost of a leaked (crashed-worker) slot self-healing more
+# slowly. period_seconds can still raise the TTL further, never lower it.
+CONCURRENCY_MIN_SAFETY_TTL_SECONDS: Final = 3600
+
+# Single-key atomic check-and-increment. Deliberately one key per script call
+# (never a batch of differently-hash-tagged keys in one call): every tag_rl
+# key carries its own self-contained {..} hash tag so unrelated buckets never
+# forcibly co-locate on the same Redis Cluster shard, which means a single Lua
+# invocation can never span more than one key's slot without risking a
+# cross-slot error. All-or-nothing across a hop's multiple atomic checks
+# (e.g. requests + concurrency checked together) is achieved in Python by
+# calling this once per key and refunding every earlier admission in the same
+# batch if a later one is rejected -- the same refund-on-rollback shape as
+# `atomic_check_and_increment_by_n` in parallel_request_limiter_v3.py, applied
+# per-key instead of per-descriptor since each key already is one hash-tag
+# group by construction.
+#
+# refresh_ttl (ARGV[4]) distinguishes the two callers of this script:
+# "requests" is an epoch-bucketed fixed window, whose TTL must be set once
+# (at first write) and never extended, or the bucket outlives the epoch it's
+# meant to reset at. "concurrency" is not windowed at all -- its TTL exists
+# purely as a crash-safety net for a reservation whose explicit release never
+# runs -- so a still-active bucket must keep pushing that TTL out on every
+# admission, or a long-lived burst of continuous traffic expires the whole
+# counter mid-flight (silently admitting past the cap, and letting a release
+# for a since-reset counter decrement an unrelated, newer cohort).
+TAG_RL_CHECK_AND_INCR_SCRIPT: Final = """
+local key = KEYS[1]
+local limit = tonumber(ARGV[1])
+local increment = tonumber(ARGV[2])
+local ttl = tonumber(ARGV[3])
+local refresh_ttl = tonumber(ARGV[4])
+local current = tonumber(redis.call('GET', key) or 0)
+if current + increment > limit then
+    return { 0, current }
+end
+local new_value = redis.call('INCRBY', key, increment)
+if ttl > 0 then
+    if refresh_ttl == 1 then
+        redis.call('EXPIRE', key, ttl)
+    else
+        local current_ttl = redis.call('TTL', key)
+        if current_ttl == -1 then
+            redis.call('EXPIRE', key, ttl)
+        end
+    end
+end
+return { 1, new_value }
+"""
+
+# Atomic decrement that never leaves a counter negative. Used both to refund
+# an earlier admission when a later key in the same batch is rejected, and to
+# release a concurrency reservation -- floors at 0 so a decrement that can't
+# be attributed to the exact reservation that caused it (see
+# `_release_keys`'s docstring) degrades to under-counting rather than a
+# negative counter that would admit unlimited requests. Floors via DEL, not
+# `SET key 0`: releasing a reservation whose key already expired makes
+# INCRBY recreate it with no TTL, and a plain SET would leave that recreated
+# key permanently in Redis (SET clears any TTL); DEL removes it outright,
+# which reads back identically to 0 everywhere this key is read (`GET key or 0`).
+TAG_RL_DECR_FLOOR_ZERO_SCRIPT: Final = """
+local key = KEYS[1]
+local delta = tonumber(ARGV[1])
+local new_value = redis.call('INCRBY', key, delta)
+if new_value < 0 then
+    redis.call('DEL', key)
+    new_value = 0
+end
+return new_value
+"""
+
+# A (tag_id, values) pair mirroring TagRateLimitScope's own fields, used to
+# fold `enabled_for`/`disabled_for` into a hashable form (a policy
+# fingerprint, or a per-deployment dedup signature) without depending on
+# TagRateLimitScope's own hashability.
+ScopeSignature: TypeAlias = tuple[str, tuple[str, ...]] | None
+
+
+def scope_signature(scope: TagRateLimitScope | None) -> ScopeSignature:
+    """Normalizes a `TagRateLimitScope` into a plain, hashable tuple, so
+    `enabled_for`/`disabled_for` can be folded into a policy fingerprint (or
+    a per-deployment dedup signature) -- two entries disagreeing on either
+    field must be treated as genuinely different policies rather than
+    merged into one bucket."""
+    return None if scope is None else (scope.tag_id, scope.values)
+
+
+def extract_identity(tags: Sequence[str], tag_id: str) -> str | None:
+    """
+    First tag matching `f"{tag_id}:"`, value after the colon. Tags starting
+    with `!` are tag-routing negation markers, not identity tags, and are
+    skipped so they can never be misread as an identity value.
+    """
+    prefix: Final = f"{tag_id}:"
+    for tag in tags:
+        if tag.startswith("!"):
+            continue
+        if tag.startswith(prefix):
+            return tag[len(prefix) :]
+    return None
+
+
+def entry_applies(entry: TagRateLimitEntry, tags: Sequence[str], key_alias: str | None, model: str | None) -> bool:
+    """
+    Applies `entry`'s own scoping fields (`enabled_for`/`disabled_for`/
+    `apply_to_key_alias`/`apply_to_models`), evaluated in this order -- deny
+    overrides allow, checked before any allowlist:
+
+      1. `disabled_for`: the gate tag (often a SECOND, independent tag, but
+         `disabled_for.tag_id` can equally be set to this entry's own
+         `tag_id` to gate on a subset of its own resolved identity) is
+         present and its value is in `disabled_for.values` -> doesn't apply.
+         Absent gate tag never triggers this -- nothing to match against a
+         denylist.
+      2. `enabled_for`: the gate tag is absent, or present but its value is
+         NOT in `enabled_for.values` -> doesn't apply. Unlike `disabled_for`,
+         absence DOES fail this check -- an allowlist gate requires an
+         explicit match, so "not tagged at all" means "not in scope".
+      3. `apply_to_models`: `model` is absent, or present but not in the
+         list -> doesn't apply. Same allowlist semantics as `enabled_for` --
+         a request with no `model` never satisfies this gate.
+      4. `apply_to_key_alias`: the calling key's own alias is absent, or
+         present but not in the list -> doesn't apply. Same allowlist
+         semantics as `enabled_for` -- a key with no alias set never
+         satisfies this gate.
+
+    An entry with none of these fields set always applies -- this is the
+    unscoped behavior every existing entry has today, unchanged.
+    """
+    if entry.disabled_for is not None:
+        disabled_gate_value: Final = extract_identity(tags, entry.disabled_for.tag_id)
+        if disabled_gate_value is not None and disabled_gate_value in entry.disabled_for.values:
+            return False
+    if entry.enabled_for is not None:
+        enabled_gate_value: Final = extract_identity(tags, entry.enabled_for.tag_id)
+        if enabled_gate_value is None or enabled_gate_value not in entry.enabled_for.values:
+            return False
+    if entry.apply_to_models is not None and model not in entry.apply_to_models:
+        return False
+    if entry.apply_to_key_alias is None:
+        return True
+    return key_alias in entry.apply_to_key_alias
+
+
+def resolve_authoritative_metadata_variable_name(
+    metadata_source: Mapping[str, object],
+) -> Literal["metadata", "litellm_metadata"]:
+    """`get_metadata_variable_name_from_kwargs` only checks key presence, which
+    misresolves both at admission time and at `async_log_success_event` time:
+    a caller can forge an empty (or merely present-but-`None`) `litellm_metadata`
+    on an ordinary request -- `kwargs["litellm_params"]` also always carries a
+    `litellm_metadata` key (typically `None`) alongside the real, populated
+    `metadata` dict for a standard (non LITELLM_METADATA_ROUTES) request -- and
+    the key-presence check always picks `litellm_metadata` in both cases,
+    silently reading no tags/identity at all and admitting the request against
+    every configured limit.
+
+    Merely requiring the value to be a non-empty dict is not enough either: a
+    caller can populate its own, unrelated keys on the non-authoritative
+    bucket (e.g. `{"litellm_metadata": {"x": 1}}` on an ordinary route), which
+    is non-empty but still not the field the proxy wrote identity into.
+    `add_litellm_data_to_request` unconditionally stamps `user_api_key_auth`
+    into whichever bucket the route actually resolved as authoritative, and
+    strips any `user_api_key_`-prefixed key a caller pre-populates on the
+    other bucket -- so requiring that marker's presence, not mere
+    truthiness, can't be forged onto the wrong side."""
+    litellm_metadata: Final = metadata_source.get("litellm_metadata")
+    if isinstance(litellm_metadata, Mapping) and "user_api_key_auth" in litellm_metadata:
+        return "litellm_metadata"
+    return "metadata"
+
+
+def extract_key_hash(request_kwargs: Mapping[str, object], metadata_variable_name: str) -> str | None:
+    """Same single-authoritative-field lookup as
+    `model_based_tag_rate_limits_hook._extract_team_id`, but for the calling
+    virtual key's hash: `LiteLLMProxyRequestSetup` sets
+    `metadata["user_api_key"]` to `user_api_key_dict.api_key`, which despite
+    the plain name is already the hashed token (see `litellm_pre_call_utils.py`).
+    """
+    active: Final = request_kwargs.get(metadata_variable_name) or EMPTY_MAPPING
+    key_hash: Final = active.get("user_api_key")
+    return key_hash if isinstance(key_hash, str) else None
+
+
+def extract_key_alias(request_kwargs: Mapping[str, object], metadata_variable_name: str) -> str | None:
+    """Same single-authoritative-field lookup as
+    `model_based_tag_rate_limits_hook._extract_team_id`, but for the calling
+    virtual key's own `key_alias`: `LiteLLMProxyRequestSetup` sets
+    `metadata["user_api_key_alias"]` to `user_api_key_dict.key_alias`
+    (see `litellm_pre_call_utils.py`)."""
+    active: Final = request_kwargs.get(metadata_variable_name) or EMPTY_MAPPING
+    key_alias: Final = active.get("user_api_key_alias")
+    return key_alias if isinstance(key_alias, str) else None
+
+
+def _active_metadata_bucket(request_kwargs: Mapping[str, object], metadata_variable_name: str) -> Mapping[str, object]:
+    """Same fallback `_get_tags_from_request_kwargs` (tag_based_routing.py)
+    already relies on: `request_kwargs` is a flat, top-level-metadata dict at
+    admission time, but `Logging.model_call_details` (what `kwargs` actually
+    is by `async_log_success_event`/`async_log_failure_event` time) never
+    carries `metadata`/`litellm_metadata` at its own top level, only nested
+    under `request_kwargs["litellm_params"]`. Checking only the top level
+    silently finds nothing at success/failure time, exactly the same failure
+    mode that lookup already had to handle."""
+    top_level: Final = request_kwargs.get(metadata_variable_name)
+    if isinstance(top_level, Mapping):
+        return top_level
+    litellm_params: Final = request_kwargs.get("litellm_params")
+    if isinstance(litellm_params, Mapping):
+        nested: Final = litellm_params.get(metadata_variable_name)
+        if isinstance(nested, Mapping):
+            return nested
+    return EMPTY_MAPPING
+
+
+def order_tags_for_identity_resolution(
+    tags: Sequence[str], request_kwargs: Mapping[str, object], metadata_variable_name: str
+) -> tuple[str, ...]:
+    """`extract_identity`/`entry_applies` both resolve a `tag_id` via
+    first-match-by-prefix. `_merge_tags` (litellm_pre_call_utils.py) appends
+    key/team/project tags only if not already present, keeping caller-supplied
+    tags first in the merged `tags` list -- so an authenticated caller could
+    submit e.g. `company_id:attacker-chosen` ahead of the calling key's real
+    `company_id:real-company` tag and have every entry scoped to `company_id`
+    resolve to the caller's own value instead of the key's. `metadata.inherited_tags`
+    is a separate, server-computed snapshot of only the tags the calling
+    key/team/project's own config contributed (see that field's docstring in
+    litellm_pre_call_utils.py), so putting it first makes a policy-backed tag
+    win over a same-prefix caller-supplied one.
+    """
+    active: Final = _active_metadata_bucket(request_kwargs, metadata_variable_name)
+    inherited_tags: Final = active.get("inherited_tags")
+    if not isinstance(inherited_tags, (list, tuple)) or not inherited_tags:
+        return tuple(tags)
+    return tuple(dict.fromkeys((*inherited_tags, *tags)))
+
+
+def fixed_length_identity(tag_value: str) -> str:
+    """
+    `tag_value` is caller-controlled (whatever follows the tag_id prefix in
+    a caller-supplied tag) with no length or content bound. Embedding it
+    directly would let a caller inflate a hook's own in-memory dict keys
+    past what `max_in_memory_cache_size` bounds (that caps item *count*, not
+    key bytes) and grow unbounded Redis keys with no cap at all. Hashing to
+    a fixed-length digest bounds a hook's own contribution to key size
+    regardless of the caller's input, while still preserving distinctness
+    (two different tag values still resolve to two different buckets).
+    """
+    return hashlib.sha256(tag_value.encode()).hexdigest()
+
+
+def policy_fingerprint(entry: TagRateLimitEntry) -> str:
+    """
+    Two entries can share a `name` and `tag_id` while genuinely disagreeing
+    on `limit`, `period_seconds`, `scope_by_key_hash`, or any of the scoping
+    fields -- both hooks already treat that as two distinct policies
+    elsewhere (model_based_tag_rate_limits_hook.py's own per-deployment
+    dedup signature is one example), so the Redis/in-memory bucket key must
+    too, or two differently-configured entries that happen to share a name
+    check and charge the identical counter. `scope_by_key_hash` specifically
+    needs its own slot here rather than relying on a hook's own `_hash_tag`
+    `key_hash`-derived suffix to carry it: that suffix is empty whenever
+    `key_hash` resolves to `None` (no virtual key on the call), which would
+    otherwise collide an unscoped entry with a key-hash-scoped one that
+    agrees on every other field. Hashed to a fixed-length digest for the
+    same reason `fixed_length_identity` hashes `tag_value`: an operator's
+    own `enabled_for`/`disabled_for`/`apply_to_key_alias`/`apply_to_models`
+    list has no length bound.
+    """
+    fingerprint_source: Final = (
+        entry.limit,
+        entry.period_seconds,
+        entry.scope_by_key_hash,
+        scope_signature(entry.enabled_for),
+        scope_signature(entry.disabled_for),
+        entry.apply_to_key_alias,
+        entry.apply_to_models,
+    )
+    return hashlib.sha256(repr(fingerprint_source).encode()).hexdigest()[:16]
+
+
+def bucket_ttl_seconds(entry: TagRateLimitEntry) -> int:
+    """Redis (and in-memory fallback) TTL for a non-concurrency bucket key.
+    `entry.key_ttl_seconds` overrides the default of period_seconds + 3600
+    when set -- see TagRateLimitEntry.key_ttl_seconds."""
+    return entry.key_ttl_seconds if entry.key_ttl_seconds is not None else entry.period_seconds + 3600
+
+
+# None => this entry shares the hook's single default cache partition
+# (matching every entry's behavior before this override existed). Otherwise
+# a value-stable signature -- not the override int alone -- so two different
+# entries that happen to choose the identical max_in_memory_cache_size don't
+# get merged into one shared partition; the same entry (same config content)
+# always resolves to the same signature across config/index rebuilds, which
+# is what keeps each hook's own `_partitions` cache from leaking a fresh
+# partition every time its configuration is re-resolved.
+# The str before the trailing int is `policy_fingerprint(entry)`: two
+# entries can share tag_id/name while genuinely disagreeing on
+# limit/period_seconds/scope_by_key_hash/enabled_for/disabled_for/
+# apply_to_key_alias/apply_to_models -- both hooks already treat that as two
+# distinct policies for bucket-key purposes, so a shared
+# max_in_memory_cache_size must not route them onto the same partition
+# either, or one entry's high-cardinality traffic can evict the other's
+# active counters from a cache neither entry asked to share.
+# max_in_memory_cache_size stays the trailing element: `partition_key[-1]`
+# reads it directly to size the partition's cache.
+PartitionKey: TypeAlias = tuple[str, str, str, int] | None
+# Grouping type for async_log_success_event's per-partition tokens/dollars
+# pipeline dispatch -- named only so the declaration fits on one line; see
+# that method for why the grouping is needed.
+PartitionOperations: TypeAlias = dict[PartitionKey, list[RedisPipelineIncrementOperation]]
+
+
+def partition_key(entry: TagRateLimitEntry) -> PartitionKey:
+    if entry.max_in_memory_cache_size is None:
+        return None
+    return (
+        entry.tag_id,
+        entry.name,
+        policy_fingerprint(entry),
+        entry.max_in_memory_cache_size,
+    )

litellm/types/router.py

@@ -4,6 +4,7 @@ litellm.Router Types - includes RouterConfig, UpdateRouterConfig, ModelInfo etc
 
 import datetime
 import enum
+import math
 from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints
@@ -137,6 +138,176 @@ def _as_utc(value: datetime.datetime | None) -> datetime.datetime | None:
     return value.astimezone(datetime.timezone.utc)
 
 
+class TagRateLimitScope(BaseModel):
+    """
+    A gate on a tag OTHER than the entry's own `tag_id` -- e.g. scoping an
+    entry to `tag_id: company_id, values: ["1032"]` so it only applies to
+    requests tagged as belonging to company 1032, independent of whichever
+    tag the entry itself keys its bucket by. See `TagRateLimitEntry.enabled_for`/
+    `disabled_for`, which are the only two fields that construct this.
+    """
+
+    tag_id: str
+    values: tuple[str, ...]
+
+    model_config = ConfigDict(frozen=True)
+
+    @model_validator(mode="after")
+    def _validate_values(self) -> "TagRateLimitScope":
+        if not self.values:
+            raise ValueError("values must be a non-empty list of strings")
+        return self
+
+    @model_validator(mode="after")
+    def _normalize_values(self) -> "TagRateLimitScope":
+        # Sorted and deduplicated: `values` is only ever used for membership
+        # tests (see _entry_applies), never order-dependent, but is also
+        # folded verbatim into the dedup signature two deployments' entries
+        # are compared by (see _scope_signature) -- an unsorted tuple would
+        # make config-order alone, not policy, decide whether two entries
+        # dedup to one shared bucket or wrongly split into two.
+        # object.__setattr__ bypasses this frozen model's own assignment
+        # guard -- returning a replacement instance from an "after" validator
+        # is silently ignored when constructing via __init__ (only takes
+        # effect via model_validate), so mutating in place is the only way
+        # this normalization reliably applies regardless of construction path.
+        object.__setattr__(self, "values", tuple(sorted(set(self.values))))  # mutable-ok: frozen before escaping
+        return self
+
+
+class TagRateLimitEntry(BaseModel):
+    name: str
+    tag_id: str = "end_user_id"
+    limit: float
+    period_seconds: int
+    scope_by_key_hash: bool = False
+    # Overrides this entry's bucket/reservation key TTL (Redis, and the
+    # in-memory fallback when Redis isn't configured). Defaults to
+    # period_seconds + 3600 when unset -- see _PROXY_ModelBasedTagRateLimitsHook._ttl_for.
+    # A high-cardinality tag_id can keep many keys alive at once; lowering
+    # this lets an operator shed them sooner without shortening
+    # period_seconds itself.
+    key_ttl_seconds: int | None = None
+    # Overrides the size of the dedicated in-memory cache partition this
+    # entry's own keys live in, when Redis isn't configured (or as a local
+    # fast-path cache when it is). Unset means this entry shares the hook's
+    # single default partition, sized by
+    # litellm.model_based_tag_rate_limits_max_in_memory_cache_size (200 if that's also
+    # unset). A high-cardinality tag_id can churn past that shared cap and
+    # evict another entry's active counters; setting this gives the entry
+    # its own dedicated partition instead.
+    max_in_memory_cache_size: int | None = None
+    # Gate this entry on a tag -- often a SECOND, independent tag (e.g.
+    # `enabled_for: {tag_id: company_id, values: ["1032"]}` to scope an
+    # override to one company's traffic), but `tag_id` can equally be set to
+    # this same entry's own `tag_id` to scope by a subset of its own
+    # resolved identity instead, without a second tag at all.
+    # `disabled_for` is checked first (deny overrides allow) when both are
+    # set. An absent gate tag never satisfies `enabled_for` (an allowlist
+    # gate requires an explicit match) but never triggers `disabled_for`
+    # either (nothing to match against a denylist).
+    enabled_for: TagRateLimitScope | None = None
+    disabled_for: TagRateLimitScope | None = None
+    # Restrict this entry to requests authenticated with one of these virtual
+    # keys' own `key_alias`. Unset (the default) means the entry applies to
+    # every request regardless of which key made it. A key with no alias set
+    # never satisfies this allowlist, same "absent gate never matches an
+    # allowlist" precedent as `enabled_for`.
+    apply_to_key_alias: tuple[str, ...] | None = None
+    # Restrict this entry to requests whose caller-facing `model` matches one
+    # of these names. Unset (the default) means the entry applies to every
+    # model. A request with no `model` field never satisfies this allowlist,
+    # same "absent gate never matches an allowlist" precedent as
+    # `apply_to_key_alias`.
+    apply_to_models: tuple[str, ...] | None = None
+
+    model_config = ConfigDict(protected_namespaces=())
+
+    @model_validator(mode="after")
+    def _validate_limit(self) -> "TagRateLimitEntry":
+        # NaN compares False against every ordering operator, so a NaN limit
+        # makes the atomic requests/concurrency check-and-increment (which
+        # rejects when the new value exceeds the limit) never reject --
+        # admitting indefinitely -- while the read-only tokens/dollars check
+        # (which admits when the current value is under the limit) never
+        # admits, rejecting every tagged request. Either outcome silently
+        # defeats the entry; reject it at config load time instead.
+        if math.isnan(self.limit):
+            raise ValueError("limit must not be NaN")
+        # Positive infinity never rejects the checks that gate this limit; negative
+        # infinity always does. Both silently defeat the entry.
+        if math.isinf(self.limit):
+            raise ValueError("limit must be finite")
+        # Zero or negative makes every check that gates this limit either always
+        # reject or never admit, silently blocking or admitting all matching traffic
+        # instead of the likely intended config.
+        if self.limit <= 0:
+            raise ValueError("limit must be a positive number")
+        return self
+
+    @model_validator(mode="after")
+    def _validate_period_seconds(self) -> "TagRateLimitEntry":
+        if self.period_seconds <= 0:
+            raise ValueError("period_seconds must be a positive integer")
+        return self
+
+    @model_validator(mode="after")
+    def _validate_key_ttl_seconds(self) -> "TagRateLimitEntry":
+        if self.key_ttl_seconds is not None and self.key_ttl_seconds <= 0:
+            raise ValueError("key_ttl_seconds must be a positive integer when set")
+        # A shorter TTL than period_seconds expires the counter before its period
+        # elapses, letting it reset early and exceed the limit.
+        if self.key_ttl_seconds is not None and self.key_ttl_seconds < self.period_seconds:
+            raise ValueError("key_ttl_seconds must be at least period_seconds when set")
+        return self
+
+    @model_validator(mode="after")
+    def _validate_max_in_memory_cache_size(self) -> "TagRateLimitEntry":
+        if self.max_in_memory_cache_size is not None and self.max_in_memory_cache_size <= 0:
+            raise ValueError("max_in_memory_cache_size must be a positive integer when set")
+        return self
+
+    @model_validator(mode="after")
+    def _validate_apply_to_key_alias(self) -> "TagRateLimitEntry":
+        if self.apply_to_key_alias is not None and not self.apply_to_key_alias:
+            raise ValueError("apply_to_key_alias must be a non-empty list of strings when set")
+        return self
+
+    @model_validator(mode="after")
+    def _normalize_apply_to_key_alias(self) -> "TagRateLimitEntry":
+        # Sorted and deduplicated for the same reason as
+        # TagRateLimitScope._normalize_values: only ever used for membership
+        # tests, but also folded verbatim into the dedup signature, where an
+        # unsorted tuple would make config-order alone decide whether two
+        # deployments' entries dedup to one shared bucket.
+        if self.apply_to_key_alias is not None:
+            self.apply_to_key_alias = tuple(sorted(set(self.apply_to_key_alias)))  # mutable-ok: frozen before escaping
+        return self
+
+    @model_validator(mode="after")
+    def _validate_apply_to_models(self) -> "TagRateLimitEntry":
+        if self.apply_to_models is not None and not self.apply_to_models:
+            raise ValueError("apply_to_models must be a non-empty list of strings when set")
+        return self
+
+    @model_validator(mode="after")
+    def _normalize_apply_to_models(self) -> "TagRateLimitEntry":
+        if self.apply_to_models is not None:
+            self.apply_to_models = tuple(sorted(set(self.apply_to_models)))  # mutable-ok: frozen before escaping
+        return self
+
+
+class TagRateLimitGroup(BaseModel):
+    limits: tuple[TagRateLimitEntry, ...] = ()
+
+
+class TagRateLimits(BaseModel):
+    token_limits: TagRateLimitGroup | None = None
+    request_limits: TagRateLimitGroup | None = None
+    dollar_limits: TagRateLimitGroup | None = None
+    concurrency_limits: TagRateLimitGroup | None = None
+
+
 class ModelInfo(MirroredPricingParams):
     id: str | None  # Allow id to be optional on input, but it will always be present as a str in the model instance
     db_model: bool = False  # used for proxy - to separate models which are stored in the db vs. config.
@@ -186,6 +357,8 @@ class ModelInfo(MirroredPricingParams):
     # router-wide default.
     enable_tag_filtering: bool | None = None
 
+    tag_rate_limits: TagRateLimits | None = None
+
     def __init__(self, id: str | int | None = None, **params) -> None:
         if id is None:
             id = str(uuid.uuid4())  # Generate a UUID if id is None or not provided

tests/test_litellm/caching/test_in_memory_cache.py

@@ -45,6 +45,39 @@ async def test_async_increment_delegates_to_locked_sync_path():
     assert cache.get_cache("counter") == 5
 
 
+async def test_async_increment_pipeline_applies_each_operations_own_ttl():
+    """
+    Bugbot finding: async_increment_pipeline dropped each operation's own
+    "ttl" field, so a counter created through it always got the cache's
+    600-second default_ttl regardless of what the caller actually configured
+    (e.g. an hourly or daily rate-limit window) -- the counter (and whatever
+    it was tracking against a limit) silently reset mid-window.
+    """
+    cache = InMemoryCache(default_ttl=600)
+    await cache.async_increment_pipeline([{"key": "long-window-counter", "increment_value": 1, "ttl": 7200}])
+
+    ttl_remaining = await cache.async_get_ttl("long-window-counter")
+    assert ttl_remaining is not None
+    assert ttl_remaining > time.time() + 600
+
+
+async def test_async_increment_pipeline_preserves_an_existing_live_ttl_on_later_increments():
+    """
+    A second increment on the same still-live counter must not reset its
+    remaining ttl back up to the full window -- only the first increment
+    (the one that actually creates the counter) should set it.
+    """
+    cache = InMemoryCache(default_ttl=600)
+    await cache.async_increment_pipeline([{"key": "counter", "increment_value": 1, "ttl": 7200}])
+    ttl_after_first = cache.ttl_dict["counter"]
+
+    await cache.async_increment_pipeline([{"key": "counter", "increment_value": 1, "ttl": 7200}])
+    ttl_after_second = cache.ttl_dict["counter"]
+
+    assert ttl_after_second == ttl_after_first
+    assert cache.get_cache("counter") == 2
+
+
 def test_in_memory_openai_obj_cache():
     from openai import OpenAI
 

tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py

@@ -0,0 +1,383 @@
+"""
+Unit tests for the primitives shared by both tag-scoped rate-limit hooks
+(`model_based_tag_rate_limits_hook.py` and `global_tag_rate_limits_hook.py`).
+
+These test the hook-independent logic in isolation: identity extraction,
+`entry_applies` scoping, and the partition/bucket-TTL key helpers. Each
+hook's own test file covers everything specific to how it wires these
+primitives into its own admission/accounting engine.
+"""
+
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    bucket_ttl_seconds,
+    entry_applies,
+    extract_identity,
+    extract_key_hash,
+    fixed_length_identity,
+    order_tags_for_identity_resolution,
+    partition_key,
+)
+from litellm.types.router import TagRateLimitEntry, TagRateLimitScope
+
+# ---------------------------------------------------------------------------
+# extract_identity
+# ---------------------------------------------------------------------------
+
+
+def test_extract_identity_matches_prefixed_tag():
+    assert extract_identity(["team_id:t1", "end_user_id:u1"], "end_user_id") == "u1"
+
+
+def test_extract_identity_returns_none_when_absent():
+    assert extract_identity(["team_id:t1"], "end_user_id") is None
+
+
+def test_extract_identity_skips_negation_tags():
+    """A `!end_user_id:u1` routing-negation marker must never be read as identity."""
+    assert extract_identity(["!end_user_id:u1"], "end_user_id") is None
+
+
+# ---------------------------------------------------------------------------
+# order_tags_for_identity_resolution -- veria-ai finding on PR #38292: a
+# caller-supplied tag must not shadow a policy-backed (key/team/project)
+# tag sharing the same tag_id prefix
+# ---------------------------------------------------------------------------
+
+
+def test_order_tags_for_identity_resolution_prefers_inherited_tag_over_caller_supplied():
+    request_kwargs = {"metadata": {"inherited_tags": ["company_id:real-company"]}}
+    tags = ["company_id:attacker-chosen", "end_user_id:u1"]
+    ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
+    assert extract_identity(ordered, "company_id") == "real-company"
+
+
+def test_order_tags_for_identity_resolution_falls_back_to_caller_tags_when_nothing_inherited():
+    request_kwargs = {"metadata": {}}
+    tags = ["end_user_id:u1"]
+    ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
+    assert extract_identity(ordered, "end_user_id") == "u1"
+
+
+def test_order_tags_for_identity_resolution_keeps_caller_only_tags_not_shadowed_by_a_different_tag_id():
+    request_kwargs = {"metadata": {"inherited_tags": ["company_id:real-company"]}}
+    tags = ["end_user_id:u1"]
+    ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
+    assert extract_identity(ordered, "end_user_id") == "u1"
+    assert extract_identity(ordered, "company_id") == "real-company"
+
+
+def test_order_tags_for_identity_resolution_deduplicates_identical_tag_present_in_both_sources():
+    request_kwargs = {"metadata": {"inherited_tags": ["company_id:real-company"]}}
+    tags = ["company_id:real-company"]
+    ordered = order_tags_for_identity_resolution(tags, request_kwargs, "metadata")
+    assert ordered.count("company_id:real-company") == 1
+
+
+# ---------------------------------------------------------------------------
+# fixed_length_identity -- tag_value is caller-controlled with no length
+# bound; a hook's own contribution to a cache key must not grow with it
+# ---------------------------------------------------------------------------
+
+
+def test_fixed_length_identity_bounds_key_contribution_regardless_of_input_size():
+    """
+    A caller can submit an arbitrarily long tag value (no length or content
+    bound is enforced upstream of either hook). Without hashing, that value
+    would go straight into an in-memory dict key (bypassing
+    max_in_memory_cache_size, which caps item *count* not key bytes) and an
+    unbounded-length Redis key (Redis has no key-count or key-size cap at
+    all here). A fixed-length digest bounds a hook's own contribution to
+    the key regardless of input size.
+    """
+    huge_value = "x" * 5_000_000
+    digest = fixed_length_identity(huge_value)
+    assert len(digest) == 64  # sha256 hex digest length, independent of input size
+
+
+def test_fixed_length_identity_preserves_distinctness():
+    """Hashing must not collapse two different tag values onto one bucket."""
+    assert fixed_length_identity("user-a") != fixed_length_identity("user-b")
+    assert fixed_length_identity("user-a") == fixed_length_identity("user-a")
+
+
+# ---------------------------------------------------------------------------
+# extract_key_hash -- must read only the one field the server actually
+# authenticates into, never fall back to the other
+# ---------------------------------------------------------------------------
+
+
+def test_extract_key_hash_ignores_a_forged_value_in_the_non_authoritative_field():
+    """
+    On a route where litellm_metadata is authoritative, the server writes
+    the real hash there and never touches metadata -- so a caller-supplied
+    metadata.user_api_key must not be read at all, let alone win.
+    """
+    request_kwargs = {
+        "metadata": {"user_api_key": "forged-by-caller"},
+        "litellm_metadata": {"user_api_key": "real-authenticated-hash"},
+    }
+    assert extract_key_hash(request_kwargs, "litellm_metadata") == "real-authenticated-hash"
+
+
+def test_extract_key_hash_reads_metadata_when_it_is_the_authoritative_field():
+    request_kwargs = {"metadata": {"user_api_key": "real-hash"}}
+    assert extract_key_hash(request_kwargs, "metadata") == "real-hash"
+
+
+# ---------------------------------------------------------------------------
+# entry_applies -- enabled_for / disabled_for / apply_to_key_alias / apply_to_models
+# ---------------------------------------------------------------------------
+
+
+def test_entry_applies_with_none_of_the_scoping_fields_set():
+    entry = TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=500, period_seconds=86400)
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
+
+
+def test_entry_applies_disabled_for_on_its_own_tag_id_excludes_a_listed_value():
+    """disabled_for's `tag_id` can be set to the entry's own tag_id, gating on
+    a subset of its own resolved identity rather than a second tag."""
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        disabled_for=TagRateLimitScope(tag_id="end_user_id", values=("u1",)),
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
+    assert entry_applies(entry, ["end_user_id:u2"], None, None) is True
+
+
+def test_entry_applies_enabled_for_on_its_own_tag_id_restricts_to_a_listed_value():
+    """enabled_for's `tag_id` can likewise be set to the entry's own tag_id,
+    admitting only a hand-picked subset of its own resolved identity."""
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        enabled_for=TagRateLimitScope(tag_id="end_user_id", values=("u2", "u3")),
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
+    assert entry_applies(entry, ["end_user_id:u2"], None, None) is True
+
+
+def test_entry_applies_matches_an_enabled_for_gate():
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
+    )
+    assert entry_applies(entry, ["end_user_id:u1", "company_id:1032"], None, None) is True
+
+
+def test_entry_applies_skips_when_enabled_for_gate_tag_is_absent():
+    """
+    enabled_for is an allowlist gate: absence of the gate tag must not
+    satisfy it, unlike disabled_for below.
+    """
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
+
+
+def test_entry_applies_skips_when_disabled_for_gate_matches():
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        disabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
+    )
+    assert entry_applies(entry, ["end_user_id:u1", "company_id:1032"], None, None) is False
+
+
+def test_entry_applies_when_disabled_for_gate_tag_is_absent():
+    """disabled_for is a denylist gate: absence of the gate tag has nothing
+    to match against, so the entry still applies."""
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        disabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
+
+
+def test_entry_applies_disabled_for_overrides_a_matching_enabled_for_gate():
+    """Deny (disabled_for) takes effect independently of whether the
+    enabled_for gate itself matched, even when both target the same tag."""
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
+        disabled_for=TagRateLimitScope(tag_id="end_user_id", values=("u1",)),
+    )
+    assert entry_applies(entry, ["end_user_id:u1", "company_id:1032"], None, None) is False
+
+
+def test_entry_applies_with_apply_to_key_alias_unset_applies_to_every_key():
+    entry = TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=500, period_seconds=86400)
+    assert entry_applies(entry, ["end_user_id:u1"], "any-key-alias", None) is True
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
+
+
+def test_entry_applies_admits_a_key_alias_on_the_allowlist():
+    entry = TagRateLimitEntry(
+        name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_key_alias=("team-a-key",)
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], "team-a-key", None) is True
+
+
+def test_entry_applies_rejects_a_key_alias_missing_from_the_allowlist():
+    entry = TagRateLimitEntry(
+        name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_key_alias=("team-a-key",)
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], "team-b-key", None) is False
+
+
+def test_entry_applies_rejects_when_key_has_no_alias_but_allowlist_is_set():
+    """apply_to_key_alias is an allowlist gate: a key with no alias at all
+    never satisfies it, same as enabled_for's absent-gate-tag semantics."""
+    entry = TagRateLimitEntry(
+        name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_key_alias=("team-a-key",)
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
+
+
+def test_entry_applies_with_apply_to_models_unset_applies_to_every_model():
+    entry = TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=500, period_seconds=86400)
+    assert entry_applies(entry, ["end_user_id:u1"], None, "opus-chain") is True
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is True
+
+
+def test_entry_applies_admits_a_model_on_the_apply_to_models_allowlist():
+    entry = TagRateLimitEntry(
+        name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_models=("opus-chain",)
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, "opus-chain") is True
+
+
+def test_entry_applies_rejects_a_model_missing_from_the_apply_to_models_allowlist():
+    entry = TagRateLimitEntry(
+        name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_models=("opus-chain",)
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, "sonnet-chain") is False
+
+
+def test_entry_applies_rejects_when_model_is_absent_but_apply_to_models_is_set():
+    """apply_to_models is an allowlist gate: a request with no model at all
+    never satisfies it, same as apply_to_key_alias's absent-key semantics."""
+    entry = TagRateLimitEntry(
+        name="daily", tag_id="end_user_id", limit=500, period_seconds=86400, apply_to_models=("opus-chain",)
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], None, None) is False
+
+
+def test_entry_applies_apply_to_models_composes_with_apply_to_key_alias():
+    """Both gates must pass: a request against the listed model but a
+    non-listed key alias must not apply, even though apply_to_models alone
+    would have admitted it."""
+    entry = TagRateLimitEntry(
+        name="daily",
+        tag_id="end_user_id",
+        limit=500,
+        period_seconds=86400,
+        apply_to_models=("opus-chain",),
+        apply_to_key_alias=("premium-key",),
+    )
+    assert entry_applies(entry, ["end_user_id:u1"], "premium-key", "opus-chain") is True
+    assert entry_applies(entry, ["end_user_id:u1"], "other-key", "opus-chain") is False
+    assert entry_applies(entry, ["end_user_id:u1"], "premium-key", "sonnet-chain") is False
+
+
+# ---------------------------------------------------------------------------
+# partition_key -- entries that share max_in_memory_cache_size but disagree
+# on any policy-fingerprinted field must never share a cache partition
+# ---------------------------------------------------------------------------
+
+
+def test_partition_key_distinguishes_entries_that_differ_only_by_scope_by_key_hash():
+    """
+    scope_by_key_hash is part of the partition-key signature: two entries
+    identical in every other field but differing only on this flag are
+    different rate limits (different bucket keys per each hook's own
+    `_hash_tag`) and must never be routed to the same cache partition.
+    """
+    unscoped = TagRateLimitEntry(
+        name="per_minute", tag_id="end_user_id", limit=5, period_seconds=60, max_in_memory_cache_size=100
+    )
+    scoped = TagRateLimitEntry(
+        name="per_minute",
+        tag_id="end_user_id",
+        limit=5,
+        period_seconds=60,
+        scope_by_key_hash=True,
+        max_in_memory_cache_size=100,
+    )
+    assert partition_key(unscoped) != partition_key(scoped)
+
+
+def test_partition_key_distinguishes_entries_that_differ_only_by_scoping_fields():
+    """
+    A plain, unscoped entry and a scoped override can legitimately share
+    name/tag_id/limit/period_seconds/scope_by_key_hash while disagreeing on
+    enabled_for/disabled_for/apply_to_key_alias/apply_to_models --
+    policy_fingerprint already treats that as two distinct policies, so a
+    shared max_in_memory_cache_size must not route them onto the same
+    in-memory partition either, or one entry's high-cardinality traffic can
+    evict the other's active counters from a cache neither entry asked to
+    share.
+    """
+    base_kwargs = {
+        "name": "daily",
+        "tag_id": "end_user_id",
+        "limit": 100,
+        "period_seconds": 86400,
+        "max_in_memory_cache_size": 50,
+    }
+    unscoped = TagRateLimitEntry(**base_kwargs)
+    enabled_for_scoped = TagRateLimitEntry(
+        **base_kwargs, enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",))
+    )
+    disabled_for_scoped = TagRateLimitEntry(
+        **base_kwargs, disabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",))
+    )
+    alias_scoped = TagRateLimitEntry(**base_kwargs, apply_to_key_alias=("premium-key",))
+    models_scoped = TagRateLimitEntry(**base_kwargs, apply_to_models=("opus-chain",))
+
+    keys = {
+        partition_key(unscoped),
+        partition_key(enabled_for_scoped),
+        partition_key(disabled_for_scoped),
+        partition_key(alias_scoped),
+        partition_key(models_scoped),
+    }
+    assert len(keys) == 5
+
+
+# ---------------------------------------------------------------------------
+# bucket_ttl_seconds -- per-tag Redis/bucket key TTL override
+# ---------------------------------------------------------------------------
+
+
+def test_bucket_ttl_seconds_defaults_to_period_plus_one_hour_when_unset():
+    entry = TagRateLimitEntry(name="per_minute", tag_id="end_user_id", limit=1, period_seconds=60)
+    assert bucket_ttl_seconds(entry) == 60 + 3600
+
+
+def test_bucket_ttl_seconds_honors_key_ttl_seconds_override():
+    entry = TagRateLimitEntry(
+        name="per_minute", tag_id="end_user_id", limit=1, period_seconds=60, key_ttl_seconds=120
+    )
+    assert bucket_ttl_seconds(entry) == 120

tests/test_litellm/proxy/test_common_request_processing.py

@@ -33,6 +33,7 @@ from litellm.proxy.common_request_processing import (
     ttft_keepalive_interval,
     _override_openai_response_model,
     _parse_event_data_for_error,
+    _release_disconnect_state_on_all_callbacks,
     _resolve_per_request_model_group_alias,
     _should_return_raw_model_name,
     _UpstreamClosingStreamingResponse,
@@ -4063,7 +4064,52 @@ class TestCancelOnDisconnect:
         llm_call.cancel()
 
         with pytest.raises(asyncio.CancelledError):
-            await _await_llm_call_cancelling_on_disconnect(request, llm_call)
+            await _await_llm_call_cancelling_on_disconnect(request, llm_call, {})
+
+    async def test_disconnect_releases_callback_state_before_499(self, monkeypatch):
+        """
+        asyncio.CancelledError is a BaseException, not an Exception, so it
+        never reaches litellm.utils.wrapper_async's own except block -- the
+        cancelled call's async_log_failure_event never fires, and the 499
+        this raises is later handled by post_call_failure_hook, a different
+        hook a CustomLogger like model_based_tag_rate_limits_hook doesn't implement. Without
+        an explicit release here, a callback that reserved per-request state
+        at admission (a concurrency slot) leaks it until that state's own
+        safety TTL. This mirrors the streaming disconnect case
+        (_finalize_streaming_generator_cleanup), just for a non-streaming
+        call cancelled via the opt-in cancel_on_disconnect flag.
+        """
+        recorder = _RecordingDisconnectHookLogger()
+        monkeypatch.setattr(litellm, "callbacks", [recorder])
+        request = self._request([{"type": "http.disconnect"}])
+        llm_call = asyncio.get_running_loop().create_future()
+
+        with pytest.raises(HTTPException) as exc_info:
+            await _await_llm_call_cancelling_on_disconnect(
+                request, llm_call, {"litellm_logging_obj": MagicMock()}
+            )
+
+        assert exc_info.value.status_code == 499
+        assert recorder.disconnect_hook_calls == 1
+
+    async def test_release_disconnect_state_calls_every_callback_including_ones_without_an_override(
+        self, monkeypatch
+    ):
+        """
+        Bugbot finding: CustomLogger never defined async_release_disconnect_state_hook
+        as an empty default, unlike every other optional hook on that base class, so
+        calling it on a callback that never overrides it (most registered callbacks)
+        raised AttributeError -- caught here, but still a real gap in the base class's
+        own contract that a genuine implementation bug would be indistinguishable from.
+        """
+        overriding = _RecordingDisconnectHookLogger()
+        bare = CustomLogger()
+        monkeypatch.setattr(litellm, "callbacks", [overriding, bare])
+
+        await _release_disconnect_state_on_all_callbacks({"litellm_call_id": "call-1"})
+
+        assert overriding.disconnect_hook_calls == 1
+        assert await bare.async_release_disconnect_state_hook({"litellm_call_id": "call-1"}) is None
 
     async def _drive_base_process_llm_request(
         self, monkeypatch, general_settings: dict, llm_call, request: Request
@@ -5450,6 +5496,129 @@ class TestPreCallWithFallbacksOnLocalRateLimit:
 
         assert processor.data["model"] == primary_model
 
+    @pytest.mark.asyncio
+    async def test_cross_model_scoped_rejection_is_not_retried_via_fallback(self):
+        """
+        veria-ai finding on PR #36541: an entry using ``apply_to_models`` to cap
+        an entire fallback chain as one unit is defeated by this exact mechanism
+        if a fallback model isn't also listed in ``apply_to_models`` -- the
+        rejection here is a deliberate "this whole chain is capped" decision,
+        not a "this one model is unhealthy" signal, so retrying against an
+        unlisted fallback silently serves a request the operator's policy meant
+        to block. ``detail["cross_model_scope"]`` is the marker
+        global_tag_rate_limits_hook sets for exactly this case; the fallback
+        handler must re-raise immediately instead of trying any fallback model.
+        """
+        from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError
+        from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+
+        primary_model = "opus-chain"
+
+        processor = ProxyBaseLLMRequestProcessing(data={"model": primary_model})
+
+        call_count = 0
+
+        async def mock_pre_call_logic(**kwargs):
+            nonlocal call_count
+            call_count += 1
+            raise ProxyRateLimitError(
+                detail={"error": "tag_rate_limit_exceeded", "cross_model_scope": True},
+                headers={"retry-after": "30"},
+            )
+
+        mock_router = MagicMock()
+        mock_router.fallbacks = [{"opus-chain": ["sonnet-chain"]}]
+
+        with patch.object(
+            processor,
+            "common_processing_pre_call_logic",
+            side_effect=mock_pre_call_logic,
+        ):
+            with pytest.raises(ProxyRateLimitError):
+                await processor._pre_call_with_fallbacks(
+                    request=MagicMock(),
+                    general_settings={},
+                    proxy_logging_obj=MagicMock(),
+                    user_api_key_dict=MagicMock(router_settings=None),
+                    version=None,
+                    proxy_config=MagicMock(),
+                    user_model=None,
+                    user_temperature=None,
+                    user_request_timeout=None,
+                    user_max_tokens=None,
+                    user_api_base=None,
+                    model=primary_model,
+                    route_type="acompletion",
+                    llm_router=mock_router,
+                )
+
+        assert call_count == 1
+        assert processor.data["model"] == primary_model
+
+    @pytest.mark.asyncio
+    async def test_cross_model_scoped_rejection_mid_chain_stops_further_fallback_attempts(self):
+        """
+        Bugbot finding: the original exception is checked for
+        cross_model_scope before the fallback loop starts, but a LATER
+        fallback attempt's own rejection was never checked the same way --
+        the loop's `except ProxyRateLimitError: continue` swallowed it and
+        moved on to the next fallback model. If a chain-wide apply_to_models
+        cap covers both the primary model and the first fallback, and a
+        second fallback model isn't covered, this let the second fallback
+        silently serve the request the cap was meant to block. The original
+        (non-scoped) rejection enters the loop normally; the FIRST fallback's
+        own rejection carries cross_model_scope=True and must stop the loop
+        immediately, never reaching the second fallback.
+        """
+        from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError
+        from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
+
+        primary_model = "opus-chain"
+
+        processor = ProxyBaseLLMRequestProcessing(data={"model": primary_model})
+
+        attempted_models: list[str] = []
+
+        async def mock_pre_call_logic(**kwargs):
+            attempted_models.append(processor.data["model"])
+            if processor.data["model"] == primary_model:
+                # Original attempt: a plain, non-scoped rejection (e.g. a
+                # per-deployment limit), not the chain-wide cap itself.
+                raise ProxyRateLimitError(detail={"error": "tag_rate_limit_exceeded"}, headers={"retry-after": "30"})
+            if processor.data["model"] == "sonnet-chain":
+                # First fallback: rejected by the SAME chain-wide cap.
+                raise ProxyRateLimitError(
+                    detail={"error": "tag_rate_limit_exceeded", "cross_model_scope": True},
+                    headers={"retry-after": "30"},
+                )
+            raise AssertionError(f"must not attempt a second fallback model: {processor.data['model']}")
+
+        mock_router = MagicMock()
+        mock_router.fallbacks = [{"opus-chain": ["sonnet-chain", "haiku-chain"]}]
+
+        with patch.object(processor, "common_processing_pre_call_logic", side_effect=mock_pre_call_logic):
+            with pytest.raises(ProxyRateLimitError) as exc_info:
+                await processor._pre_call_with_fallbacks(
+                    request=MagicMock(),
+                    general_settings={},
+                    proxy_logging_obj=MagicMock(),
+                    user_api_key_dict=MagicMock(router_settings=None),
+                    version=None,
+                    proxy_config=MagicMock(),
+                    user_model=None,
+                    user_temperature=None,
+                    user_request_timeout=None,
+                    user_max_tokens=None,
+                    user_api_base=None,
+                    model=primary_model,
+                    route_type="acompletion",
+                    llm_router=mock_router,
+                )
+
+        assert attempted_models == [primary_model, "sonnet-chain"]
+        assert exc_info.value.detail.get("cross_model_scope") is True
+        assert processor.data["model"] == primary_model
+
     @pytest.mark.asyncio
     async def test_real_parallel_request_limiter_model_tpm_limit_triggers_fallback(self):
         """
@@ -5596,6 +5765,15 @@ class _RecordingSuccessLogger(CustomLogger):
         self.success_events.append({"kwargs": kwargs, "response_obj": response_obj})
 
 
+class _RecordingDisconnectHookLogger(CustomLogger):
+    def __init__(self):
+        super().__init__()
+        self.disconnect_hook_calls = 0
+
+    async def async_release_disconnect_state_hook(self, request_data: dict) -> None:
+        self.disconnect_hook_calls += 1
+
+
 class TestStreamingClientDisconnectBilling:
     """
     A client disconnect throws GeneratorExit into the proxy streaming
@@ -5990,6 +6168,64 @@ class TestStreamingClientDisconnectBilling:
         assert usage.prompt_tokens_details is not None
         assert usage.prompt_tokens_details.cached_tokens == 7
 
+    @pytest.mark.asyncio
+    async def test_disconnect_without_billable_chunks_releases_callback_state(self, monkeypatch):
+        """
+        A callback that reserves per-request state outside of the success/failure
+        logging callbacks (e.g. a concurrency slot admitted before the first
+        chunk) would otherwise leak it on a disconnect with nothing to bill,
+        since neither logging callback ever fires for it. The disconnect
+        cleanup must give every registered callback a chance to release such
+        state via async_release_disconnect_state_hook.
+        """
+        import types
+
+        response = await self._start_partial_stream()
+        empty_response = types.SimpleNamespace(chunks=[], messages=None)
+        recorder = _RecordingDisconnectHookLogger()
+        monkeypatch.setattr(litellm, "callbacks", [recorder])
+        await ProxyBaseLLMRequestProcessing._finalize_streaming_generator_cleanup(
+            request=None,
+            request_data={"litellm_logging_obj": response.logging_obj},
+            response=empty_response,
+            stream_completed=False,
+            client_disconnected=True,
+            user_api_key_dict=MagicMock(),
+            proxy_logging_obj=types.SimpleNamespace(
+                _arelease_max_parallel_requests_on_disconnect=AsyncMock(),
+            ),
+        )
+
+        assert recorder.disconnect_hook_calls == 1
+
+    @pytest.mark.asyncio
+    async def test_disconnect_billing_skips_callback_disconnect_hook(self, monkeypatch):
+        """
+        When a disconnect-time success event already fired (partial billing
+        dispatched it), that event's own async_log_success_event already ran
+        for every registered callback. The disconnect hook must not also run
+        in that case, so a callback with idempotent-but-not-free release logic
+        does not do redundant work on every disconnect.
+        """
+        import types
+
+        recorder = _RecordingDisconnectHookLogger()
+        monkeypatch.setattr(litellm, "callbacks", [recorder])
+        response = await self._start_partial_stream()
+        await ProxyBaseLLMRequestProcessing._finalize_streaming_generator_cleanup(
+            request=None,
+            request_data={"litellm_logging_obj": response.logging_obj},
+            response=response,
+            stream_completed=False,
+            client_disconnected=True,
+            user_api_key_dict=MagicMock(),
+            proxy_logging_obj=types.SimpleNamespace(
+                _arelease_max_parallel_requests_on_disconnect=AsyncMock(),
+            ),
+        )
+
+        assert recorder.disconnect_hook_calls == 0
+
 
 def _apply_stream_usage_tracking(
     data: dict,

tests/test_litellm/types/test_router.py

@@ -5,6 +5,7 @@ from litellm.types.router import (
     Deployment,
     LiteLLM_Params,
     ModelInfo,
+    TagRateLimitEntry,
 )
 from litellm.types.utils import CustomPricingLiteLLMParams, MirroredPricingParams
 
@@ -89,3 +90,47 @@ def test_pricing_strings_are_coerced_to_float():
 def test_invalid_pricing_is_rejected():
     with pytest.raises(ValueError, match='validation error for ModelInfo'):
         ModelInfo(id="x", input_cost_per_token="free")
+
+
+# litellm/types/router.py is imported by plain SDK users, not just the proxy, so a
+# TagRateLimitEntry validation error should read as a generic config-validation
+# message and not describe the proxy rate-limit hook's internal admission mechanics.
+_INTERNAL_ENFORCEMENT_JARGON = (
+    "admission",
+    "tagged request",
+    "tagged traffic",
+    "check-and-increment",
+    "read-only",
+    "atomic",
+    "window rolls over",
+)
+
+
+def _assert_message_has_no_internal_jargon(excinfo: pytest.ExceptionInfo) -> None:
+    message = str(excinfo.value).lower()
+    leaked = [term for term in _INTERNAL_ENFORCEMENT_JARGON if term in message]
+    assert not leaked, f"validation message leaked internal enforcement jargon: {leaked}"
+
+
+def test_limit_infinite_rejected_without_internal_enforcement_jargon():
+    with pytest.raises(ValueError, match="validation error for TagRateLimitEntry") as excinfo:
+        TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=float("inf"), period_seconds=60)
+    _assert_message_has_no_internal_jargon(excinfo)
+
+
+def test_limit_non_positive_rejected_without_internal_enforcement_jargon():
+    with pytest.raises(ValueError, match="validation error for TagRateLimitEntry") as excinfo:
+        TagRateLimitEntry(name="daily", tag_id="end_user_id", limit=0, period_seconds=60)
+    _assert_message_has_no_internal_jargon(excinfo)
+
+
+def test_key_ttl_seconds_shorter_than_period_rejected_without_internal_enforcement_jargon():
+    with pytest.raises(ValueError, match="validation error for TagRateLimitEntry") as excinfo:
+        TagRateLimitEntry(
+            name="daily",
+            tag_id="end_user_id",
+            limit=500,
+            period_seconds=86400,
+            key_ttl_seconds=60,
+        )
+    _assert_message_has_no_internal_jargon(excinfo)

type-discipline-budget.json

@@ -27,7 +27,7 @@
     "limit": 0
   },
   "LIT010": {
-    "limit": 16616
+    "limit": 16615
   },
   "LIT011": {
     "limit": 5583

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

@@ -35115,6 +35115,64 @@ export interface components {
             /** Tpm Limit */
             tpm_limit?: number | null;
         };
+        /** TagRateLimitEntry */
+        TagRateLimitEntry: {
+            /** Apply To Key Alias */
+            apply_to_key_alias?: string[] | null;
+            /** Apply To Models */
+            apply_to_models?: string[] | null;
+            disabled_for?: components["schemas"]["TagRateLimitScope"] | null;
+            enabled_for?: components["schemas"]["TagRateLimitScope"] | null;
+            /** Key Ttl Seconds */
+            key_ttl_seconds?: number | null;
+            /** Limit */
+            limit: number;
+            /** Max In Memory Cache Size */
+            max_in_memory_cache_size?: number | null;
+            /** Name */
+            name: string;
+            /** Period Seconds */
+            period_seconds: number;
+            /**
+             * Scope By Key Hash
+             * @default false
+             */
+            scope_by_key_hash: boolean;
+            /**
+             * Tag Id
+             * @default end_user_id
+             */
+            tag_id: string;
+        };
+        /** TagRateLimitGroup */
+        TagRateLimitGroup: {
+            /**
+             * Limits
+             * @default []
+             */
+            limits: components["schemas"]["TagRateLimitEntry"][];
+        };
+        /**
+         * TagRateLimitScope
+         * @description A gate on a tag OTHER than the entry's own `tag_id` -- e.g. scoping an
+         *     entry to `tag_id: company_id, values: ["1032"]` so it only applies to
+         *     requests tagged as belonging to company 1032, independent of whichever
+         *     tag the entry itself keys its bucket by. See `TagRateLimitEntry.enabled_for`/
+         *     `disabled_for`, which are the only two fields that construct this.
+         */
+        TagRateLimitScope: {
+            /** Tag Id */
+            tag_id: string;
+            /** Values */
+            values: string[];
+        };
+        /** TagRateLimits */
+        TagRateLimits: {
+            concurrency_limits?: components["schemas"]["TagRateLimitGroup"] | null;
+            dollar_limits?: components["schemas"]["TagRateLimitGroup"] | null;
+            request_limits?: components["schemas"]["TagRateLimitGroup"] | null;
+            token_limits?: components["schemas"]["TagRateLimitGroup"] | null;
+        };
         /**
          * TagSummaryMetrics
          * @description Summary metrics for a tag
@@ -37890,6 +37948,7 @@ export interface components {
             ptu_effective_from?: string | null;
             /** Ptu Effective To */
             ptu_effective_to?: string | null;
+            tag_rate_limits?: components["schemas"]["TagRateLimits"] | null;
             /** Team Id */
             team_id?: string | null;
             /** Team Public Model Name */