From 0d5425c3fc13a15a52e152ae02347f2a5860f200 Mon Sep 17 00:00:00 2001
From: Deepanshu <deepanshu.lulla@alpha-sense.com>
Date: Sat, 5 Sep 2026 07:11:22 -0400
Subject: [PATCH 1/7] feat(types): add tag rate limit config types + fix
 InMemoryCache pipeline ttl

Add TagRateLimitScope/TagRateLimitEntry/TagRateLimitGroup/TagRateLimits to
litellm/types/router.py and a ModelInfo.tag_rate_limits field. Pure config
types with no runtime behavior yet; a later PR wires up the enforcement
hooks that consume them.

Also fix InMemoryCache.async_increment_pipeline, which dropped each
pipeline operation's own ttl and always fell back to the cache's default
ttl, silently resetting long-window counters early. Independent bug fix,
found while building the tag rate limiting feature but unrelated to it.

First of a resplit of the tag-based rate limiting feature into much
smaller PRs, per maintainer feedback that ~250 LOC PRs review well and
the prior 3-way split (#38289, #38292, #38347, #36541) was still too
large; those four PRs are now closed in favor of this resplit.
---
 litellm/caching/in_memory_cache.py            |  10 +-
 litellm/types/router.py                       | 173 ++++++++++++++++++
 .../caching/test_in_memory_cache.py           |  33 ++++
 tests/test_litellm/types/test_router.py       |  45 +++++
 ui/litellm-dashboard/src/lib/http/schema.d.ts |  59 ++++++
 5 files changed, 319 insertions(+), 1 deletion(-)

diff --git a/litellm/caching/in_memory_cache.py b/litellm/caching/in_memory_cache.py
index 38a9966f9f9..7edf4672963 100644
--- a/litellm/caching/in_memory_cache.py
+++ b/litellm/caching/in_memory_cache.py
@@ -248,7 +248,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
 
diff --git a/litellm/types/router.py b/litellm/types/router.py
index 360f7b20ec7..3cdcbaa8a4f 100644
--- a/litellm/types/router.py
+++ b/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, Sequence
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints
@@ -160,6 +161,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.
@@ -214,6 +385,8 @@ class ModelInfo(MirroredPricingParams):
     # in the spend log row's metadata. Set it on every deployment of the group.
     internal_router_model: 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
diff --git a/tests/test_litellm/caching/test_in_memory_cache.py b/tests/test_litellm/caching/test_in_memory_cache.py
index 85e8308ae91..1e03159ca9a 100644
--- a/tests/test_litellm/caching/test_in_memory_cache.py
+++ b/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
 
diff --git a/tests/test_litellm/types/test_router.py b/tests/test_litellm/types/test_router.py
index 7fcb76b638f..3427fbd2f89 100644
--- a/tests/test_litellm/types/test_router.py
+++ b/tests/test_litellm/types/test_router.py
@@ -9,6 +9,7 @@ from litellm.types.router import (
     GenericLiteLLMParams,
     LiteLLM_Params,
     ModelInfo,
+    TagRateLimitEntry,
 )
 from litellm.types.utils import CustomPricingLiteLLMParams, MirroredPricingParams
 
@@ -146,3 +147,47 @@ def test_aws_session_tags_round_trip_as_sts_shaped_pairs():
 def test_aws_session_tags_reject_shapes_sts_would_refuse(aws_session_tags):
     with pytest.raises(ValidationError, match="aws_session_tags"):
         LiteLLM_Params(model="bedrock/anthropic.claude-opus-5", aws_session_tags=aws_session_tags)
+
+
+# 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)
diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts
index 5c4661ab357..0c21fb78cce 100644
--- a/ui/litellm-dashboard/src/lib/http/schema.d.ts
+++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts
@@ -36671,6 +36671,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
@@ -39661,6 +39719,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 */

From 146ebe5094bd0422598b3a2f943d1d06322d2f80 Mon Sep 17 00:00:00 2001
From: Deepanshu <deepanshu.lulla@alpha-sense.com>
Date: Sat, 5 Sep 2026 07:22:04 -0400
Subject: [PATCH 2/7] refactor: trim excessive source comments in tag rate
 limit types

Greptile flagged the new TagRateLimitEntry/TagRateLimitScope docstrings
and the InMemoryCache ttl forwarding comment as excessive per the
repo's source-comment rule. Condensed each to one concise line while
keeping the non-obvious behavior they document.
---
 litellm/caching/in_memory_cache.py            |  7 +-
 litellm/types/router.py                       | 78 ++-----------------
 ui/litellm-dashboard/src/lib/http/schema.d.ts |  6 +-
 3 files changed, 10 insertions(+), 81 deletions(-)

diff --git a/litellm/caching/in_memory_cache.py b/litellm/caching/in_memory_cache.py
index 7edf4672963..58a0c14fbeb 100644
--- a/litellm/caching/in_memory_cache.py
+++ b/litellm/caching/in_memory_cache.py
@@ -248,12 +248,7 @@ class InMemoryCache(BaseCache):
     ) -> list[float] | None:
         results: Final = []
         for increment in increment_list:
-            # 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.
+            # forward each op's own ttl; allow_ttl_override leaves an already-live ttl untouched
             result = await self.async_increment(
                 increment["key"], increment["increment_value"], ttl=increment.get("ttl"), **kwargs
             )
diff --git a/litellm/types/router.py b/litellm/types/router.py
index 3cdcbaa8a4f..6da67a2c3d8 100644
--- a/litellm/types/router.py
+++ b/litellm/types/router.py
@@ -162,13 +162,7 @@ def _as_utc(value: datetime.datetime | None) -> datetime.datetime | None:
 
 
 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.
-    """
+    """A gate on a tag other than the entry's own `tag_id`, used by `TagRateLimitEntry.enabled_for`/`disabled_for`."""
 
     tag_id: str
     values: tuple[str, ...]
@@ -183,17 +177,7 @@ class TagRateLimitScope(BaseModel):
 
     @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.
+        # sorted+deduped so dedup-signature comparisons aren't order-sensitive; __setattr__ works around frozen=True
         object.__setattr__(self, "values", tuple(sorted(set(self.values))))  # mutable-ok: frozen before escaping
         return self
 
@@ -204,66 +188,25 @@ class TagRateLimitEntry(BaseModel):
     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.
+    # overrides the default bucket ttl (period_seconds + 3600); see _PROXY_ModelBasedTagRateLimitsHook._ttl_for
     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.
+    # overrides the shared litellm.model_based_tag_rate_limits_max_in_memory_cache_size (200) with a dedicated partition
     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).
+    # gate this entry on another tag; disabled_for wins over enabled_for; an absent tag never matches either allowlist
     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.
+        # NaN compares False against every ordering operator, silently defeating whichever check gates this limit
         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
@@ -278,8 +221,7 @@ class TagRateLimitEntry(BaseModel):
     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.
+        # a shorter ttl than period_seconds resets the counter early, letting it 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
@@ -298,11 +240,7 @@ class TagRateLimitEntry(BaseModel):
 
     @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.
+        # same reason as TagRateLimitScope._normalize_values
         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
diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts
index 0c21fb78cce..94b37901fc7 100644
--- a/ui/litellm-dashboard/src/lib/http/schema.d.ts
+++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts
@@ -36710,11 +36710,7 @@ export interface components {
         };
         /**
          * 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.
+         * @description A gate on a tag other than the entry's own `tag_id`, used by `TagRateLimitEntry.enabled_for`/`disabled_for`.
          */
         TagRateLimitScope: {
             /** Tag Id */

From cf4b80b24a2014a76a5df842bb841d9c6a74376c Mon Sep 17 00:00:00 2001
From: Deepanshu <deepanshu.lulla@alpha-sense.com>
Date: Sat, 5 Sep 2026 07:36:18 -0400
Subject: [PATCH 3/7] test: cover every TagRateLimit* validator and normalizer

Codecov flagged 57% patch coverage on the new config types: only 3 of
TagRateLimitEntry's validators had tests, and TagRateLimitScope,
TagRateLimitGroup, TagRateLimits, and ModelInfo.tag_rate_limits had
none at all. Added a test for each remaining validator/normalizer
(NaN limit, non-positive period/ttl/cache-size, empty allowlists,
sort+dedup normalization, frozen scope) and for constructing the
group/aggregate types. Spot-checked kill power on the NaN check, the
sort+dedup normalizer, and the frozen guard by reverting each one and
confirming its test fails.
---
 tests/test_litellm/types/test_router.py | 100 +++++++++++++++++++++++-
 1 file changed, 99 insertions(+), 1 deletion(-)

diff --git a/tests/test_litellm/types/test_router.py b/tests/test_litellm/types/test_router.py
index 3427fbd2f89..cec79ebf9d1 100644
--- a/tests/test_litellm/types/test_router.py
+++ b/tests/test_litellm/types/test_router.py
@@ -10,6 +10,9 @@ from litellm.types.router import (
     LiteLLM_Params,
     ModelInfo,
     TagRateLimitEntry,
+    TagRateLimitGroup,
+    TagRateLimits,
+    TagRateLimitScope,
 )
 from litellm.types.utils import CustomPricingLiteLLMParams, MirroredPricingParams
 
@@ -92,7 +95,7 @@ def test_pricing_strings_are_coerced_to_float():
 
 
 def test_invalid_pricing_is_rejected():
-    with pytest.raises(ValueError, match='validation error for ModelInfo'):
+    with pytest.raises(ValueError, match="validation error for ModelInfo"):
         ModelInfo(id="x", input_cost_per_token="free")
 
 
@@ -191,3 +194,98 @@ def test_key_ttl_seconds_shorter_than_period_rejected_without_internal_enforceme
             key_ttl_seconds=60,
         )
     _assert_message_has_no_internal_jargon(excinfo)
+
+
+def test_limit_nan_rejected():
+    """NaN compares False against every ordering operator, so it would otherwise slip
+    past a `limit <= 0` style guard and silently defeat whichever check gates it."""
+    with pytest.raises(ValueError, match="limit must not be NaN"):
+        TagRateLimitEntry(name="daily", limit=float("nan"), period_seconds=60)
+
+
+def test_period_seconds_non_positive_rejected():
+    with pytest.raises(ValueError, match="period_seconds must be a positive integer"):
+        TagRateLimitEntry(name="daily", limit=10, period_seconds=0)
+
+
+def test_key_ttl_seconds_non_positive_rejected():
+    with pytest.raises(ValueError, match="key_ttl_seconds must be a positive integer when set"):
+        TagRateLimitEntry(name="daily", limit=10, period_seconds=60, key_ttl_seconds=0)
+
+
+def test_max_in_memory_cache_size_non_positive_rejected():
+    with pytest.raises(ValueError, match="max_in_memory_cache_size must be a positive integer"):
+        TagRateLimitEntry(name="daily", limit=10, period_seconds=60, max_in_memory_cache_size=0)
+
+
+def test_apply_to_key_alias_empty_list_rejected():
+    with pytest.raises(ValueError, match="apply_to_key_alias must be a non-empty list"):
+        TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_key_alias=())
+
+
+def test_apply_to_key_alias_sorted_and_deduped():
+    entry = TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_key_alias=("b", "a", "a"))
+    assert entry.apply_to_key_alias == ("a", "b")
+
+
+def test_apply_to_models_empty_list_rejected():
+    with pytest.raises(ValueError, match="apply_to_models must be a non-empty list"):
+        TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_models=())
+
+
+def test_apply_to_models_sorted_and_deduped():
+    entry = TagRateLimitEntry(name="daily", limit=10, period_seconds=60, apply_to_models=("gpt-4o", "claude", "claude"))
+    assert entry.apply_to_models == ("claude", "gpt-4o")
+
+
+def test_scope_values_empty_list_rejected():
+    with pytest.raises(ValueError, match="values must be a non-empty list"):
+        TagRateLimitScope(tag_id="company_id", values=())
+
+
+def test_scope_values_sorted_and_deduped():
+    scope = TagRateLimitScope(tag_id="company_id", values=("1032", "1001", "1001"))
+    assert scope.values == ("1001", "1032")
+
+
+def test_scope_is_frozen():
+    scope = TagRateLimitScope(tag_id="company_id", values=("1032",))
+    with pytest.raises(ValidationError):
+        scope.tag_id = "other_tag"
+
+
+def test_entry_accepts_enabled_for_and_disabled_for_scopes():
+    entry = TagRateLimitEntry(
+        name="daily",
+        limit=10,
+        period_seconds=60,
+        enabled_for=TagRateLimitScope(tag_id="company_id", values=("1032",)),
+        disabled_for=TagRateLimitScope(tag_id="company_id", values=("9999",)),
+    )
+    assert entry.enabled_for.values == ("1032",)
+    assert entry.disabled_for.values == ("9999",)
+
+
+def test_group_defaults_to_no_limits():
+    assert TagRateLimitGroup().limits == ()
+
+
+def test_rate_limits_group_holds_multiple_entries():
+    entries = (
+        TagRateLimitEntry(name="daily", limit=10, period_seconds=60),
+        TagRateLimitEntry(name="weekly", limit=100, period_seconds=604800),
+    )
+    group = TagRateLimitGroup(limits=entries)
+    assert group.limits == entries
+
+
+def test_model_info_tag_rate_limits_defaults_to_none():
+    assert ModelInfo(id="x").tag_rate_limits is None
+
+
+def test_model_info_accepts_tag_rate_limits():
+    limits = TagRateLimits(
+        token_limits=TagRateLimitGroup(limits=(TagRateLimitEntry(name="daily", limit=10, period_seconds=60),))
+    )
+    info = ModelInfo(id="x", tag_rate_limits=limits)
+    assert info.tag_rate_limits.token_limits.limits[0].name == "daily"

From 1dfe1d968c8edae4299d4a595086caac2a0ff712 Mon Sep 17 00:00:00 2001
From: Deepanshu <deepanshu.lulla@alpha-sense.com>
Date: Tue, 8 Sep 2026 16:49:09 -0400
Subject: [PATCH 4/7] fix: reject empty tag_id on TagRateLimitScope and
 TagRateLimitEntry

Greptile flagged this as P1 on the stacked #39902: both tag
identifiers accepted an empty string, and downstream identity lookup
builds a key off tag_id, so an empty value would silently search for
a bare colon prefix and never match instead of erroring at config
load time. Rejects it at construction, matching the existing
non-empty guard on TagRateLimitScope.values. Verified the new tests
fail without the fix and pass with it.
---
 litellm/types/router.py                 | 12 ++++++++++++
 tests/test_litellm/types/test_router.py | 14 ++++++++++++++
 2 files changed, 26 insertions(+)

diff --git a/litellm/types/router.py b/litellm/types/router.py
index 6da67a2c3d8..7d32b4f1938 100644
--- a/litellm/types/router.py
+++ b/litellm/types/router.py
@@ -169,6 +169,12 @@ class TagRateLimitScope(BaseModel):
 
     model_config = ConfigDict(frozen=True)
 
+    @model_validator(mode="after")
+    def _validate_tag_id(self) -> "TagRateLimitScope":
+        if not self.tag_id:
+            raise ValueError("tag_id must be a non-empty string")
+        return self
+
     @model_validator(mode="after")
     def _validate_values(self) -> "TagRateLimitScope":
         if not self.values:
@@ -200,6 +206,12 @@ class TagRateLimitEntry(BaseModel):
 
     model_config = ConfigDict(protected_namespaces=())
 
+    @model_validator(mode="after")
+    def _validate_tag_id(self) -> "TagRateLimitEntry":
+        if not self.tag_id:
+            raise ValueError("tag_id must be a non-empty string")
+        return self
+
     @model_validator(mode="after")
     def _validate_limit(self) -> "TagRateLimitEntry":
         # NaN compares False against every ordering operator, silently defeating whichever check gates this limit
diff --git a/tests/test_litellm/types/test_router.py b/tests/test_litellm/types/test_router.py
index cec79ebf9d1..d67a84d9d3c 100644
--- a/tests/test_litellm/types/test_router.py
+++ b/tests/test_litellm/types/test_router.py
@@ -238,6 +238,13 @@ def test_apply_to_models_sorted_and_deduped():
     assert entry.apply_to_models == ("claude", "gpt-4o")
 
 
+def test_scope_tag_id_empty_string_rejected():
+    """An empty tag_id makes identity lookup search for a bare `:` prefix, silently
+    never matching instead of erroring at config load time (Greptile P1 on #39902)."""
+    with pytest.raises(ValueError, match="tag_id must be a non-empty string"):
+        TagRateLimitScope(tag_id="", values=("1032",))
+
+
 def test_scope_values_empty_list_rejected():
     with pytest.raises(ValueError, match="values must be a non-empty list"):
         TagRateLimitScope(tag_id="company_id", values=())
@@ -254,6 +261,13 @@ def test_scope_is_frozen():
         scope.tag_id = "other_tag"
 
 
+def test_entry_tag_id_empty_string_rejected():
+    """Same silent-no-match failure mode as TagRateLimitScope.tag_id (Greptile P1 on
+    #39902); the default is non-empty, but an explicit override could still be empty."""
+    with pytest.raises(ValueError, match="tag_id must be a non-empty string"):
+        TagRateLimitEntry(name="daily", tag_id="", limit=10, period_seconds=60)
+
+
 def test_entry_accepts_enabled_for_and_disabled_for_scopes():
     entry = TagRateLimitEntry(
         name="daily",

From 74c2dd3e225031f4cc57a85b3ade910e1613988f Mon Sep 17 00:00:00 2001
From: Deepanshu <deepanshu.lulla@alpha-sense.com>
Date: Sat, 5 Sep 2026 08:44:11 -0400
Subject: [PATCH 5/7] feat(proxy): add shared tag rate-limit helper module

Adds tag_rate_limits_shared.py: Lua scripts for atomic bucket
admission/release, bucket-key hashing and policy-fingerprint primitives,
cache partitioning, and identity resolution / security-hardening helpers
both tag-scoped rate-limit hooks will share. Nothing consumes this module
yet; that starts in a later PR in this resplit.

PR 2 of a resplit of the tag-based rate limiting feature into much smaller
PRs. Stacked on #39900 (litellm_tag_rate_limit_types) for the
TagRateLimitScope/TagRateLimitEntry types this module depends on.
---
 litellm/proxy/hooks/tag_rate_limits_shared.py | 401 ++++++++++++++++
 .../hooks/test_tag_rate_limits_shared.py      | 431 ++++++++++++++++++
 2 files changed, 832 insertions(+)
 create mode 100644 litellm/proxy/hooks/tag_rate_limits_shared.py
 create mode 100644 tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py

diff --git a/litellm/proxy/hooks/tag_rate_limits_shared.py b/litellm/proxy/hooks/tag_rate_limits_shared.py
new file mode 100644
index 00000000000..a21523c7885
--- /dev/null
+++ b/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") if isinstance(active, Mapping) else None
+    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") if isinstance(active, Mapping) else None
+    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,
+    )
diff --git a/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py b/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
new file mode 100644
index 00000000000..2e4954cd030
--- /dev/null
+++ b/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
@@ -0,0 +1,431 @@
+"""
+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_alias,
+    extract_key_hash,
+    fixed_length_identity,
+    order_tags_for_identity_resolution,
+    partition_key,
+    resolve_authoritative_metadata_variable_name,
+)
+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"
+
+
+def test_extract_key_hash_ignores_a_non_mapping_authoritative_field():
+    """Bugbot finding: metadata can arrive as an unparsed JSON string (see
+    apply_client_tag_policy_pre_auth's own docstring on multipart/extra_body
+    routes); a truthy non-Mapping must not reach .get() and crash."""
+    request_kwargs = {"metadata": '{"user_api_key": "forged"}'}
+    assert extract_key_hash(request_kwargs, "metadata") is None
+
+
+def test_extract_key_alias_ignores_a_non_mapping_authoritative_field():
+    request_kwargs = {"metadata": '{"user_api_key_alias": "forged"}'}
+    assert extract_key_alias(request_kwargs, "metadata") is None
+
+
+# ---------------------------------------------------------------------------
+# resolve_authoritative_metadata_variable_name -- Veria AI finding: a
+# caller-supplied, non-empty litellm_metadata must not be selected over
+# metadata, the field the proxy actually wrote authenticated tags into
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_authoritative_metadata_variable_name_ignores_caller_forged_non_empty_litellm_metadata():
+    """A caller-supplied litellm_metadata with unrelated content (no
+    user_api_key_auth marker) must not be picked over metadata, even though
+    it is a non-empty dict."""
+    metadata_source = {"litellm_metadata": {"marker": True}}
+    assert resolve_authoritative_metadata_variable_name(metadata_source) == "metadata"
+
+
+def test_resolve_authoritative_metadata_variable_name_selects_litellm_metadata_when_server_written():
+    """A genuinely server-populated litellm_metadata (LITELLM_METADATA_ROUTES)
+    always carries the user_api_key_auth marker stamped by
+    add_user_api_key_auth_to_request_metadata."""
+    metadata_source = {"litellm_metadata": {"user_api_key_auth": object(), "tags": ["team_id:t1"]}}
+    assert resolve_authoritative_metadata_variable_name(metadata_source) == "litellm_metadata"
+
+
+def test_resolve_authoritative_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_absent():
+    assert resolve_authoritative_metadata_variable_name({}) == "metadata"
+
+
+def test_resolve_authoritative_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_none():
+    assert resolve_authoritative_metadata_variable_name({"litellm_metadata": None}) == "metadata"
+
+
+def test_resolve_authoritative_metadata_variable_name_defaults_to_metadata_when_litellm_metadata_empty():
+    assert resolve_authoritative_metadata_variable_name({"litellm_metadata": {}}) == "metadata"
+
+
+# ---------------------------------------------------------------------------
+# 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

From 38ec9600644696760d3561277982e2872439e499 Mon Sep 17 00:00:00 2001
From: Deepanshu <deepanshu.lulla@alpha-sense.com>
Date: Sat, 5 Sep 2026 09:05:16 -0400
Subject: [PATCH 6/7] fix(proxy): fix key-hash/alias extraction at log time,
 trim comments, add Lua script tests

Greptile/Bugbot findings from the first review round:
- extract_key_hash/extract_key_alias only checked the top-level metadata
  bucket, unlike order_tags_for_identity_resolution's _active_metadata_bucket
  helper. By async_log_success_event/async_log_failure_event time, metadata
  is only nested under litellm_params, so both silently returned None there,
  diverging from the value read at admission time. Fixed by routing both
  through _active_metadata_bucket, with regression tests for the nested case.
- Condensed the module's docstrings/comments to one concise line each,
  keeping the non-obvious behavior they document, matching the trimming
  already applied to litellm/types/router.py on the stacked PR #39900.
- Added real Redis execution coverage for both Lua scripts (admission,
  rejection, ttl refresh semantics, decrement floor) against a throwaway
  local redis-server, since fakeredis has no EVALSHA support without the
  optional lupa dependency.
---
 litellm/proxy/hooks/tag_rate_limits_shared.py | 292 ++++--------------
 .../hooks/test_tag_rate_limits_shared.py      | 146 +++++++++
 2 files changed, 209 insertions(+), 229 deletions(-)

diff --git a/litellm/proxy/hooks/tag_rate_limits_shared.py b/litellm/proxy/hooks/tag_rate_limits_shared.py
index a21523c7885..75ae67dd120 100644
--- a/litellm/proxy/hooks/tag_rate_limits_shared.py
+++ b/litellm/proxy/hooks/tag_rate_limits_shared.py
@@ -1,20 +1,7 @@
-"""
-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.
-"""
+"""Primitives shared by both tag-scoped rate-limit hooks (model_based_tag_rate_limits_hook.py,
+global_tag_rate_limits_hook.py): identity/scope extraction, policy fingerprinting, bucket-key
+hashing, and cache-partitioning. Every name here is a genuine public export (no leading
+underscore); each hook imports what it needs aliased back to its own historical private name."""
 
 import asyncio
 import hashlib
@@ -29,11 +16,8 @@ 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.
+# requests/concurrency admit via atomic check-and-increment; tokens/dollars are only known
+# after the response, so they stay a read-then-account-on-success check with an unavoidable race.
 ATOMIC_UNITS: Final[frozenset[LimitUnit]] = frozenset({"requests", "concurrency"})
 
 UNIT_TO_GROUP_FIELD: Final[Mapping[LimitUnit, str]] = MappingProxyType(
@@ -53,56 +37,22 @@ UNIT_TO_RATE_LIMIT_TYPE: Final[Mapping[LimitUnit, RateLimitType]] = MappingProxy
     }
 )
 
-# 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.
+# shared read-only fallback for an absent/None mapping, so call sites don't build a fresh {}
 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.
+# holds a strong reference to every fire-and-forget background task (concurrency release,
+# token/dollar accounting) so the event loop's weak-ref-only tracking can't garbage-collect
+# one mid-execution; each task's own completion callback discards its entry
 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.
+# floor for a concurrency reservation's self-heal ttl, regardless of period_seconds, so an
+# expiring-while-in-flight reservation can't silently admit past the limit
 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).
+# single-key atomic check-and-increment (one key per call: every tag_rl key carries its own
+# {..} hash tag, so a multi-key call could span shards and cross-slot error). refresh_ttl
+# (ARGV[4]) distinguishes callers: "requests" is a fixed window whose ttl is set once and never
+# extended; "concurrency" isn't windowed, so its crash-safety ttl must refresh on every admission.
 TAG_RL_CHECK_AND_INCR_SCRIPT: Final = """
 local key = KEYS[1]
 local limit = tonumber(ARGV[1])
@@ -127,16 +77,8 @@ 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`).
+# atomic decrement floored at 0 (refund or concurrency release); floors via DEL rather than
+# `SET key 0`, since a release on an already-expired key would otherwise recreate it with no ttl
 TAG_RL_DECR_FLOOR_ZERO_SCRIPT: Final = """
 local key = KEYS[1]
 local delta = tonumber(ARGV[1])
@@ -148,28 +90,19 @@ 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.
+# a (tag_id, values) pair mirroring TagRateLimitScope, used to fold enabled_for/disabled_for
+# into a hashable form 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."""
+    """Normalizes a TagRateLimitScope into a hashable tuple for policy fingerprinting/dedup."""
     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.
-    """
+    """First tag matching `f"{tag_id}:"`, value after the colon. `!`-prefixed tag-routing
+    negation markers are skipped so they're never misread as an identity value."""
     prefix: Final = f"{tag_id}:"
     for tag in tags:
         if tag.startswith("!"):
@@ -180,32 +113,10 @@ def extract_identity(tags: Sequence[str], tag_id: str) -> str | 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.
-    """
+    """Applies entry's own scoping fields in order (deny before allow): disabled_for excludes a
+    matching value; enabled_for requires a matching value (absence fails, unlike disabled_for);
+    apply_to_models and apply_to_key_alias are allowlists an absent model/alias never satisfies.
+    An entry with none of these set always applies."""
     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:
@@ -224,63 +135,18 @@ def entry_applies(entry: TagRateLimitEntry, tags: Sequence[str], key_alias: str
 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."""
+    """Picks metadata vs litellm_metadata by the unforgeable `user_api_key_auth` marker
+    `add_litellm_data_to_request` stamps into whichever bucket is actually authoritative --
+    key presence or truthiness alone can be forged by a caller onto the wrong bucket."""
     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") if isinstance(active, Mapping) else None
-    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") if isinstance(active, Mapping) else None
-    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."""
+    """request_kwargs carries metadata at its own top level at admission time, but only nested
+    under litellm_params by async_log_success_event/async_log_failure_event time; checks both."""
     top_level: Final = request_kwargs.get(metadata_variable_name)
     if isinstance(top_level, Mapping):
         return top_level
@@ -292,21 +158,28 @@ def _active_metadata_bucket(request_kwargs: Mapping[str, object], metadata_varia
     return EMPTY_MAPPING
 
 
+def extract_key_hash(request_kwargs: Mapping[str, object], metadata_variable_name: str) -> str | None:
+    """Reads the calling virtual key's hash from the authoritative metadata bucket (nested under
+    litellm_params by log time, same as _active_metadata_bucket's other callers); `metadata["user_api_key"]`
+    is already the hashed token despite the plain name."""
+    active: Final = _active_metadata_bucket(request_kwargs, metadata_variable_name)
+    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:
+    """Reads the calling virtual key's own key_alias from the authoritative metadata bucket."""
+    active: Final = _active_metadata_bucket(request_kwargs, metadata_variable_name)
+    key_alias: Final = active.get("user_api_key_alias")
+    return key_alias if isinstance(key_alias, str) else None
+
+
 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.
-    """
+    """Puts server-computed `metadata.inherited_tags` ahead of caller-supplied tags, so a caller
+    can't submit e.g. `company_id:attacker-chosen` and shadow the calling key's real, same-prefix
+    tag; extract_identity/entry_applies both resolve tag_id via first-match-by-prefix."""
     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:
@@ -315,37 +188,16 @@ def order_tags_for_identity_resolution(
 
 
 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).
-    """
+    """Hashes a caller-controlled, unbounded-length tag value to a fixed-length digest, bounding
+    a hook's own contribution to an in-memory cache key or Redis key regardless of input size."""
     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.
-    """
+    """Fixed-length digest folding every policy-distinguishing field of an entry (limit,
+    period_seconds, scope_by_key_hash, enabled_for/disabled_for, apply_to_key_alias/models) into
+    one bucket-key signature, so two differently-configured entries sharing a name never share
+    a counter."""
     fingerprint_source: Final = (
         entry.limit,
         entry.period_seconds,
@@ -359,34 +211,16 @@ def policy_fingerprint(entry: TagRateLimitEntry) -> str:
 
 
 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."""
+    """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."""
     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.
+# None => this entry shares the hook's single default cache partition. Otherwise a value-stable
+# signature (policy_fingerprint(entry), not the override int alone) so two entries sharing a
+# max_in_memory_cache_size don't merge into one partition; max_in_memory_cache_size stays the
+# trailing element since `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]]
 
 
diff --git a/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py b/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
index 2e4954cd030..efe3b530e8d 100644
--- a/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
+++ b/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
@@ -8,7 +8,17 @@ hook's own test file covers everything specific to how it wires these
 primitives into its own admission/accounting engine.
 """
 
+import shutil
+import socket
+import subprocess
+import time
+from collections.abc import Iterator
+
+import pytest
+
 from litellm.proxy.hooks.tag_rate_limits_shared import (
+    TAG_RL_CHECK_AND_INCR_SCRIPT,
+    TAG_RL_DECR_FLOOR_ZERO_SCRIPT,
     bucket_ttl_seconds,
     entry_applies,
     extract_identity,
@@ -139,6 +149,21 @@ def test_extract_key_alias_ignores_a_non_mapping_authoritative_field():
     assert extract_key_alias(request_kwargs, "metadata") is None
 
 
+def test_extract_key_hash_finds_metadata_nested_under_litellm_params_at_log_time():
+    """Bugbot finding: by async_log_success_event/async_log_failure_event time,
+    kwargs only carries metadata nested under litellm_params (see
+    _active_metadata_bucket's own docstring), not at request_kwargs' own top
+    level -- extract_key_hash must find it there too, or a key-hash-scoped
+    bucket's accounting silently reads a different key than admission did."""
+    request_kwargs = {"litellm_params": {"metadata": {"user_api_key": "real-hash"}}}
+    assert extract_key_hash(request_kwargs, "metadata") == "real-hash"
+
+
+def test_extract_key_alias_finds_metadata_nested_under_litellm_params_at_log_time():
+    request_kwargs = {"litellm_params": {"metadata": {"user_api_key_alias": "real-alias"}}}
+    assert extract_key_alias(request_kwargs, "metadata") == "real-alias"
+
+
 # ---------------------------------------------------------------------------
 # resolve_authoritative_metadata_variable_name -- Veria AI finding: a
 # caller-supplied, non-empty litellm_metadata must not be selected over
@@ -429,3 +454,124 @@ def test_bucket_ttl_seconds_defaults_to_period_plus_one_hour_when_unset():
 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
+
+
+# ---------------------------------------------------------------------------
+# TAG_RL_CHECK_AND_INCR_SCRIPT / TAG_RL_DECR_FLOOR_ZERO_SCRIPT -- fakeredis has
+# no EVALSHA support without the optional lupa dependency, so these run
+# against a throwaway local redis-server instead (same idiom as
+# test_batch_enqueued_tokens.py's test_redis_lua_path_full_lifecycle).
+# ---------------------------------------------------------------------------
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+@pytest.fixture
+def redis_port() -> Iterator[int]:
+    if shutil.which("redis-server") is None:
+        pytest.skip("requires a local redis-server binary to exercise the Lua script path")
+    port = _free_port()
+    proc = subprocess.Popen(
+        ["redis-server", "--port", str(port), "--save", "", "--appendonly", "no"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                sock.settimeout(0.2)
+                if sock.connect_ex(("127.0.0.1", port)) == 0:
+                    break
+        else:
+            proc.terminate()
+            pytest.skip("local redis-server did not become ready in time")
+        yield port
+    finally:
+        proc.terminate()
+        proc.wait(timeout=5)
+
+
+def _register(port: int, script: str):
+    from litellm.caching.redis_cache import RedisCache
+
+    return RedisCache(host="127.0.0.1", port=port).async_register_script(script)
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_admits_under_limit_and_sets_ttl(redis_port):
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    admitted, new_value = await run(keys=["bucket1"], args=[5, 1, 60, 0])
+    assert (admitted, new_value) == (1, 1)
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_rejects_over_limit_without_incrementing(redis_port):
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    await run(keys=["bucket2"], args=[1, 1, 60, 0])
+    rejected, current = await run(keys=["bucket2"], args=[1, 1, 60, 0])
+    assert (rejected, current) == (0, 1)
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_requests_ttl_is_set_once_and_not_refreshed(redis_port):
+    """refresh_ttl=0 (the `requests` fixed-window semantics): the epoch-bucketed
+    TTL must be set on the first write and left alone after, or the bucket
+    outlives the epoch it's meant to reset at."""
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    await run(keys=["bucket3"], args=[5, 1, 100, 0])
+    from litellm.caching.redis_cache import RedisCache
+
+    raw = RedisCache(host="127.0.0.1", port=redis_port)
+    client = raw.init_async_client()
+    first_ttl = await client.ttl("bucket3")
+    await run(keys=["bucket3"], args=[5, 1, 5, 0])
+    second_ttl = await client.ttl("bucket3")
+    assert first_ttl > 5
+    assert second_ttl > 5  # unchanged by the second call's much shorter ttl arg
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_concurrency_ttl_refreshes_on_every_admission(redis_port):
+    """refresh_ttl=1 (the `concurrency` reservation semantics): every admission
+    must push the crash-safety TTL back out, or a long-lived burst of traffic
+    expires the whole counter mid-flight."""
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    await run(keys=["bucket4"], args=[5, 1, 5, 1])
+    await run(keys=["bucket4"], args=[5, 1, 100, 1])
+    from litellm.caching.redis_cache import RedisCache
+
+    raw = RedisCache(host="127.0.0.1", port=redis_port)
+    client = raw.init_async_client()
+    ttl = await client.ttl("bucket4")
+    assert ttl > 5
+
+
+@pytest.mark.asyncio
+async def test_decr_floor_zero_floors_at_zero_and_deletes_the_key(redis_port):
+    from litellm.caching.redis_cache import RedisCache
+
+    run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
+    await run_incr(keys=["bucket5"], args=[5, 1, 60, 1])
+
+    floored = await run_decr(keys=["bucket5"], args=[-5])
+    assert floored == 0
+
+    raw = RedisCache(host="127.0.0.1", port=redis_port)
+    client = raw.init_async_client()
+    assert await client.get("bucket5") is None  # floored via DEL, not a TTL-less `SET 0`
+
+
+@pytest.mark.asyncio
+async def test_decr_floor_zero_decrements_normally_when_result_stays_non_negative(redis_port):
+    run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
+    await run_incr(keys=["bucket6"], args=[5, 3, 60, 1])
+
+    remaining = await run_decr(keys=["bucket6"], args=[-1])
+    assert remaining == 2

From cc7050d24466d6e894f317da5ce1df4b277c7659 Mon Sep 17 00:00:00 2001
From: Deepanshu <deepanshu.lulla@alpha-sense.com>
Date: Sat, 5 Sep 2026 09:17:33 -0400
Subject: [PATCH 7/7] test: move Lua script Redis tests out of the mock-only
 test_litellm tree

Greptile finding: tests/test_litellm/readme.md documents that directory as
mocked-tests-only, but the six new Lua script tests contacted a real
redis-server subprocess. Moved them verbatim to
tests/local_testing/test_tag_rate_limits_shared_redis.py, which already
hosts other real-Redis tests (test_redis_batch_optimizations.py). No
behavior change; tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
is mock-only again.
---
 .../test_tag_rate_limits_shared_redis.py      | 128 +++++++++++++++++
 .../hooks/test_tag_rate_limits_shared.py      | 131 ------------------
 2 files changed, 128 insertions(+), 131 deletions(-)
 create mode 100644 tests/local_testing/test_tag_rate_limits_shared_redis.py

diff --git a/tests/local_testing/test_tag_rate_limits_shared_redis.py b/tests/local_testing/test_tag_rate_limits_shared_redis.py
new file mode 100644
index 00000000000..8c29cd288d7
--- /dev/null
+++ b/tests/local_testing/test_tag_rate_limits_shared_redis.py
@@ -0,0 +1,128 @@
+"""
+Real-Redis execution coverage for tag_rate_limits_shared's two Lua scripts.
+
+Lives outside tests/test_litellm/ (which can only contain mocked tests, see
+tests/test_litellm/readme.md) because fakeredis has no EVALSHA support without
+the optional lupa dependency; these run against a throwaway local redis-server
+instead (same idiom as tests/test_litellm/proxy/hooks/test_batch_enqueued_tokens.py's
+test_redis_lua_path_full_lifecycle).
+"""
+
+import shutil
+import socket
+import subprocess
+import time
+from collections.abc import Iterator
+
+import pytest
+
+from litellm.caching.redis_cache import RedisCache
+from litellm.proxy.hooks.tag_rate_limits_shared import (
+    TAG_RL_CHECK_AND_INCR_SCRIPT,
+    TAG_RL_DECR_FLOOR_ZERO_SCRIPT,
+)
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+@pytest.fixture
+def redis_port() -> Iterator[int]:
+    if shutil.which("redis-server") is None:
+        pytest.skip("requires a local redis-server binary to exercise the Lua script path")
+    port = _free_port()
+    proc = subprocess.Popen(
+        ["redis-server", "--port", str(port), "--save", "", "--appendonly", "no"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        deadline = time.monotonic() + 5
+        while time.monotonic() < deadline:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                sock.settimeout(0.2)
+                if sock.connect_ex(("127.0.0.1", port)) == 0:
+                    break
+        else:
+            proc.terminate()
+            pytest.skip("local redis-server did not become ready in time")
+        yield port
+    finally:
+        proc.terminate()
+        proc.wait(timeout=5)
+
+
+def _register(port: int, script: str):
+    return RedisCache(host="127.0.0.1", port=port).async_register_script(script)
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_admits_under_limit_and_sets_ttl(redis_port):
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    admitted, new_value = await run(keys=["bucket1"], args=[5, 1, 60, 0])
+    assert (admitted, new_value) == (1, 1)
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_rejects_over_limit_without_incrementing(redis_port):
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    await run(keys=["bucket2"], args=[1, 1, 60, 0])
+    rejected, current = await run(keys=["bucket2"], args=[1, 1, 60, 0])
+    assert (rejected, current) == (0, 1)
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_requests_ttl_is_set_once_and_not_refreshed(redis_port):
+    """refresh_ttl=0 (the `requests` fixed-window semantics): the epoch-bucketed
+    TTL must be set on the first write and left alone after, or the bucket
+    outlives the epoch it's meant to reset at."""
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    await run(keys=["bucket3"], args=[5, 1, 100, 0])
+    raw = RedisCache(host="127.0.0.1", port=redis_port)
+    client = raw.init_async_client()
+    first_ttl = await client.ttl("bucket3")
+    await run(keys=["bucket3"], args=[5, 1, 5, 0])
+    second_ttl = await client.ttl("bucket3")
+    assert first_ttl > 5
+    assert second_ttl > 5  # unchanged by the second call's much shorter ttl arg
+
+
+@pytest.mark.asyncio
+async def test_check_and_incr_concurrency_ttl_refreshes_on_every_admission(redis_port):
+    """refresh_ttl=1 (the `concurrency` reservation semantics): every admission
+    must push the crash-safety TTL back out, or a long-lived burst of traffic
+    expires the whole counter mid-flight."""
+    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    await run(keys=["bucket4"], args=[5, 1, 5, 1])
+    await run(keys=["bucket4"], args=[5, 1, 100, 1])
+    raw = RedisCache(host="127.0.0.1", port=redis_port)
+    client = raw.init_async_client()
+    ttl = await client.ttl("bucket4")
+    assert ttl > 5
+
+
+@pytest.mark.asyncio
+async def test_decr_floor_zero_floors_at_zero_and_deletes_the_key(redis_port):
+    run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
+    await run_incr(keys=["bucket5"], args=[5, 1, 60, 1])
+
+    floored = await run_decr(keys=["bucket5"], args=[-5])
+    assert floored == 0
+
+    raw = RedisCache(host="127.0.0.1", port=redis_port)
+    client = raw.init_async_client()
+    assert await client.get("bucket5") is None  # floored via DEL, not a TTL-less `SET 0`
+
+
+@pytest.mark.asyncio
+async def test_decr_floor_zero_decrements_normally_when_result_stays_non_negative(redis_port):
+    run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
+    run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
+    await run_incr(keys=["bucket6"], args=[5, 3, 60, 1])
+
+    remaining = await run_decr(keys=["bucket6"], args=[-1])
+    assert remaining == 2
diff --git a/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py b/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
index efe3b530e8d..a727d38e2eb 100644
--- a/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
+++ b/tests/test_litellm/proxy/hooks/test_tag_rate_limits_shared.py
@@ -8,17 +8,7 @@ hook's own test file covers everything specific to how it wires these
 primitives into its own admission/accounting engine.
 """
 
-import shutil
-import socket
-import subprocess
-import time
-from collections.abc import Iterator
-
-import pytest
-
 from litellm.proxy.hooks.tag_rate_limits_shared import (
-    TAG_RL_CHECK_AND_INCR_SCRIPT,
-    TAG_RL_DECR_FLOOR_ZERO_SCRIPT,
     bucket_ttl_seconds,
     entry_applies,
     extract_identity,
@@ -454,124 +444,3 @@ def test_bucket_ttl_seconds_defaults_to_period_plus_one_hour_when_unset():
 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
-
-
-# ---------------------------------------------------------------------------
-# TAG_RL_CHECK_AND_INCR_SCRIPT / TAG_RL_DECR_FLOOR_ZERO_SCRIPT -- fakeredis has
-# no EVALSHA support without the optional lupa dependency, so these run
-# against a throwaway local redis-server instead (same idiom as
-# test_batch_enqueued_tokens.py's test_redis_lua_path_full_lifecycle).
-# ---------------------------------------------------------------------------
-
-
-def _free_port() -> int:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-        sock.bind(("127.0.0.1", 0))
-        return sock.getsockname()[1]
-
-
-@pytest.fixture
-def redis_port() -> Iterator[int]:
-    if shutil.which("redis-server") is None:
-        pytest.skip("requires a local redis-server binary to exercise the Lua script path")
-    port = _free_port()
-    proc = subprocess.Popen(
-        ["redis-server", "--port", str(port), "--save", "", "--appendonly", "no"],
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-    )
-    try:
-        deadline = time.monotonic() + 5
-        while time.monotonic() < deadline:
-            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-                sock.settimeout(0.2)
-                if sock.connect_ex(("127.0.0.1", port)) == 0:
-                    break
-        else:
-            proc.terminate()
-            pytest.skip("local redis-server did not become ready in time")
-        yield port
-    finally:
-        proc.terminate()
-        proc.wait(timeout=5)
-
-
-def _register(port: int, script: str):
-    from litellm.caching.redis_cache import RedisCache
-
-    return RedisCache(host="127.0.0.1", port=port).async_register_script(script)
-
-
-@pytest.mark.asyncio
-async def test_check_and_incr_admits_under_limit_and_sets_ttl(redis_port):
-    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
-    admitted, new_value = await run(keys=["bucket1"], args=[5, 1, 60, 0])
-    assert (admitted, new_value) == (1, 1)
-
-
-@pytest.mark.asyncio
-async def test_check_and_incr_rejects_over_limit_without_incrementing(redis_port):
-    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
-    await run(keys=["bucket2"], args=[1, 1, 60, 0])
-    rejected, current = await run(keys=["bucket2"], args=[1, 1, 60, 0])
-    assert (rejected, current) == (0, 1)
-
-
-@pytest.mark.asyncio
-async def test_check_and_incr_requests_ttl_is_set_once_and_not_refreshed(redis_port):
-    """refresh_ttl=0 (the `requests` fixed-window semantics): the epoch-bucketed
-    TTL must be set on the first write and left alone after, or the bucket
-    outlives the epoch it's meant to reset at."""
-    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
-    await run(keys=["bucket3"], args=[5, 1, 100, 0])
-    from litellm.caching.redis_cache import RedisCache
-
-    raw = RedisCache(host="127.0.0.1", port=redis_port)
-    client = raw.init_async_client()
-    first_ttl = await client.ttl("bucket3")
-    await run(keys=["bucket3"], args=[5, 1, 5, 0])
-    second_ttl = await client.ttl("bucket3")
-    assert first_ttl > 5
-    assert second_ttl > 5  # unchanged by the second call's much shorter ttl arg
-
-
-@pytest.mark.asyncio
-async def test_check_and_incr_concurrency_ttl_refreshes_on_every_admission(redis_port):
-    """refresh_ttl=1 (the `concurrency` reservation semantics): every admission
-    must push the crash-safety TTL back out, or a long-lived burst of traffic
-    expires the whole counter mid-flight."""
-    run = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
-    await run(keys=["bucket4"], args=[5, 1, 5, 1])
-    await run(keys=["bucket4"], args=[5, 1, 100, 1])
-    from litellm.caching.redis_cache import RedisCache
-
-    raw = RedisCache(host="127.0.0.1", port=redis_port)
-    client = raw.init_async_client()
-    ttl = await client.ttl("bucket4")
-    assert ttl > 5
-
-
-@pytest.mark.asyncio
-async def test_decr_floor_zero_floors_at_zero_and_deletes_the_key(redis_port):
-    from litellm.caching.redis_cache import RedisCache
-
-    run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
-    run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
-    await run_incr(keys=["bucket5"], args=[5, 1, 60, 1])
-
-    floored = await run_decr(keys=["bucket5"], args=[-5])
-    assert floored == 0
-
-    raw = RedisCache(host="127.0.0.1", port=redis_port)
-    client = raw.init_async_client()
-    assert await client.get("bucket5") is None  # floored via DEL, not a TTL-less `SET 0`
-
-
-@pytest.mark.asyncio
-async def test_decr_floor_zero_decrements_normally_when_result_stays_non_negative(redis_port):
-    run_incr = _register(redis_port, TAG_RL_CHECK_AND_INCR_SCRIPT)
-    run_decr = _register(redis_port, TAG_RL_DECR_FLOOR_ZERO_SCRIPT)
-    await run_incr(keys=["bucket6"], args=[5, 3, 60, 1])
-
-    remaining = await run_decr(keys=["bucket6"], args=[-1])
-    assert remaining == 2