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

@@ -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
 

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

tests/test_litellm/caching/test_in_memory_cache.py

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

tests/test_litellm/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)

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 */