fix(rate-limiting): fix success-event identity ordering, add failure-hook release, drop tag disclosure

- Pass the correctly-nested litellm_params object (not raw model_call_details
  kwargs) to order_tags_for_identity_resolution in async_log_success_event.
  kwargs has no top-level metadata at that point, so the previous call was a
  silent no-op: a forged caller tag still won at accounting time even after
  the admission-time fix. Confirmed via a direct repro against the real
  Logging pipeline, and rewrote the existing regression test's kwargs shape
  to match production instead of a shape that happened to hide the bug.
- Add async_post_call_failure_hook, releasing a reservation from an earlier
  admission attempt when _pre_call_with_fallbacks exhausts every fallback
  and re-raises without ever running the real LLM call -- the only other
  release paths (success/failure/disconnect) are tied to that call, which
  never happens.
- Drop tag_value from the client-facing rejection detail: it can resolve
  from inherited_tags (server-assigned key/team/project metadata), and
  echoing it back would disclose that identity to the caller who got
  rejected.

Each fix has a regression test confirmed to fail on the pre-fix code.

litellm/proxy/hooks/global_tag_rate_limits_hook.py

@@ -504,7 +504,11 @@ class _PROXY_GlobalTagRateLimitsHook(  # pyright: ignore[reportUnusedClass]  # o
                 "error": "tag_rate_limit_exceeded",
                 "type": unit,
                 "tag_id": entry.tag_id,
-                "tag_value": tag_value,
+                # tag_value deliberately excluded: it can resolve from
+                # inherited_tags (server-assigned key/team/project metadata),
+                # and echoing it back would disclose that identity to the
+                # caller. verbose_proxy_logger.debug above still logs it
+                # server-side for observability.
                 "limit_name": entry.name,
                 "limit": entry.limit,
                 "period_seconds": entry.period_seconds,
@@ -632,26 +636,44 @@ class _PROXY_GlobalTagRateLimitsHook(  # pyright: ignore[reportUnusedClass]  # o
         self._record_admitted_model(stash, model, renewal_allowed)
         return data
 
-    async def async_release_disconnect_state_hook(self, request_data: Mapping[str, object]) -> None:
-        stash: Final = _stash_for_call(_call_id_from_kwargs(request_data))
+    async def _release_pending_for_call_id(self, request_kwargs: Mapping[str, object]) -> None:
+        stash: Final = _stash_for_call(_call_id_from_kwargs(request_kwargs))
         if stash is None or not stash.pending_concurrency_keys:
             return
         release_keys: Final = tuple(stash.pending_concurrency_keys)
         stash.pending_concurrency_keys.clear()
         await self._release_keys(release_keys)
 
+    async def async_release_disconnect_state_hook(self, request_data: Mapping[str, object]) -> None:
+        await self._release_pending_for_call_id(request_data)
+
+    async def async_post_call_failure_hook(
+        self,
+        request_data: dict,  # mutable-ok: must match CustomLogger.async_post_call_failure_hook's own base signature exactly
+        original_exception: Exception,
+        user_api_key_dict: UserAPIKeyAuth,
+        traceback_str: str | None = None,
+    ) -> None:
+        """
+        A request that never reaches Router (every fallback model also
+        rejected, or none configured) never runs the actual LLM call, so
+        neither async_log_success_event nor async_log_failure_event -- both
+        tied to that call's own wrapper -- ever fires for it. This is the
+        only remaining release path for a reservation from an earlier,
+        successful admission attempt in the same _pre_call_with_fallbacks
+        chain. litellm_call_id survives proxy/utils.py's own stripping here
+        (only litellm_logging_obj is popped), so the same ContextVar-based
+        stash lookup as the other release hooks still works.
+        """
+        await self._release_pending_for_call_id(request_data)
+
     async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:
         # Always release regardless of which hook raised: this hook's own
         # rejection never reserves a slot, so pending_concurrency_keys is
         # already empty in that case and the check below no-ops; a rejection
         # from model_based_tag_rate_limits_hook (same error marker) can still
         # land after this hook already reserved its own slot.
-        stash: Final = _stash_for_call(_call_id_from_kwargs(kwargs))
-        if stash is None or not stash.pending_concurrency_keys:
-            return
-        release_keys: Final = tuple(stash.pending_concurrency_keys)
-        stash.pending_concurrency_keys.clear()
-        await self._release_keys(release_keys)
+        await self._release_pending_for_call_id(kwargs)
 
     async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:
         stash: Final = _stash_for_call(_call_id_from_kwargs(kwargs))
@@ -681,7 +703,7 @@ class _PROXY_GlobalTagRateLimitsHook(  # pyright: ignore[reportUnusedClass]  # o
 
         tags: Final = _order_tags_for_identity_resolution(
             _get_tags_from_request_kwargs(kwargs, metadata_variable_name=metadata_variable_name),
-            kwargs,
+            litellm_params_for_metadata,
             metadata_variable_name,
         )
         if not tags:

tests/test_litellm/proxy/hooks/test_global_tag_rate_limits_hook.py

@@ -966,6 +966,54 @@ async def test_concurrency_reservation_released_on_disconnect(time_controller, m
     assert result is not None
 
 
+@pytest.mark.asyncio
+async def test_concurrency_reservation_released_when_every_fallback_is_exhausted(time_controller, monkeypatch):
+    """
+    When _pre_call_with_fallbacks exhausts every fallback model (another
+    hook rejects each one) and re-raises, the request never reaches the
+    real LLM call -- neither async_log_success_event nor
+    async_log_failure_event, both tied to that call's own wrapper, ever
+    fires. async_post_call_failure_hook is the only remaining release path
+    for a reservation this hook already admitted earlier in that same
+    fallback chain.
+    """
+    monkeypatch.setattr(
+        litellm,
+        "global_tag_rate_limits",
+        {
+            "concurrency_limits": {
+                "limits": [{"name": "conc", "tag_id": "end_user_id", "limit": 1, "period_seconds": 60}]
+            }
+        },
+    )
+    hook = _make_hook(time_controller)
+
+    # This hook admits and reserves the only slot.
+    await hook.async_pre_call_hook(
+        user_api_key_dict=_key(),
+        cache=DualCache(),
+        data={**_data(["end_user_id:u1"], call_id="call-1"), "model": "model-a"},
+        call_type="completion",
+    )
+    # _pre_call_with_fallbacks eventually gives up (every fallback rejected
+    # by some other hook) and reports the failure via post_call_failure_hook.
+    await hook.async_post_call_failure_hook(
+        request_data={"litellm_call_id": "call-1"},
+        original_exception=ProxyRateLimitError(
+            detail={"error": "some_other_hooks_limit"}, headers={}, rate_limit_type=None
+        ),
+        user_api_key_dict=_key(),
+    )
+
+    result = await hook.async_pre_call_hook(
+        user_api_key_dict=_key(),
+        cache=DualCache(),
+        data={**_data(["end_user_id:u1"], call_id="call-2"), "model": "model-a"},
+        call_type="completion",
+    )
+    assert result is not None
+
+
 @pytest.mark.asyncio
 async def test_concurrency_reservation_released_when_disconnect_runs_in_a_forked_task(time_controller, monkeypatch):
     """
@@ -1279,11 +1327,18 @@ async def test_success_accounting_also_resolves_identity_from_the_policy_backed_
     )
     hook = _make_hook(time_controller)
 
+    # kwargs at async_log_success_event time is Logging.model_call_details:
+    # metadata/inherited_tags live nested under kwargs["litellm_params"],
+    # never at the top level -- see
+    # test_log_success_event_accounts_when_litellm_params_carries_a_null_litellm_metadata_key
+    # for the same shape.
     kwargs = {
         "litellm_call_id": "attack-1",
-        "metadata": {
-            "tags": ["company_id:attacker-chosen", "company_id:real-company"],
-            "inherited_tags": ["company_id:real-company"],
+        "litellm_params": {
+            "metadata": {
+                "tags": ["company_id:attacker-chosen", "company_id:real-company"],
+                "inherited_tags": ["company_id:real-company"],
+            },
         },
         "standard_logging_object": {"total_tokens": 0, "response_cost": 20.0},
     }
@@ -1302,3 +1357,46 @@ async def test_success_accounting_also_resolves_identity_from_the_policy_backed_
             },
             call_type="completion",
         )
+
+
+@pytest.mark.asyncio
+async def test_rejection_detail_does_not_disclose_the_resolved_tag_value(time_controller, monkeypatch):
+    """
+    tag_value can resolve from inherited_tags (server-assigned key/team/
+    project metadata via order_tags_for_identity_resolution), so echoing it
+    back in the client-facing 429 detail would disclose that identity to
+    the caller who triggered the rejection.
+    """
+    monkeypatch.setattr(
+        litellm,
+        "global_tag_rate_limits",
+        {
+            "request_limits": {
+                "limits": [{"name": "per-company", "tag_id": "company_id", "limit": 1, "period_seconds": 86400}]
+            }
+        },
+    )
+    hook = _make_hook(time_controller)
+    key = _key()
+
+    data = {
+        "litellm_call_id": "call-1",
+        "metadata": {"tags": ["company_id:secret-internal-name"], "inherited_tags": ["company_id:secret-internal-name"]},
+    }
+    await hook.async_pre_call_hook(user_api_key_dict=key, cache=DualCache(), data=data, call_type="completion")
+
+    with pytest.raises(ProxyRateLimitError) as exc_info:
+        await hook.async_pre_call_hook(
+            user_api_key_dict=key,
+            cache=DualCache(),
+            data={
+                "litellm_call_id": "call-2",
+                "metadata": {
+                    "tags": ["company_id:secret-internal-name"],
+                    "inherited_tags": ["company_id:secret-internal-name"],
+                },
+            },
+            call_type="completion",
+        )
+
+    assert "tag_value" not in exc_info.value.detail