feat(rate-limit): name the limit that rejected a request in a rate_limited_by header on the 429 (#51228)

* feat(rate-limit): name the limit that rejected a request in a rate_limited_by header on the 429

* test(rate-limit): expect rate_limited_by in the exact-headers assertion

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/hooks/parallel_request_limiter_v3.py

@@ -3347,6 +3347,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
                     headers={
                         "retry-after": str(self.window_size),
                         "rate_limit_type": str(status["rate_limit_type"]),
+                        "rate_limited_by": descriptor_key,
                         "reset_at": reset_time_formatted,
                     },
                     rate_limit_type=map_v3_rate_limit_type(status["rate_limit_type"]),

tests/integration/management/test_customer_model_limits.py

@@ -53,3 +53,21 @@ def test_customer_per_model_rpm_limit_is_shared_across_keys_and_isolated_per_cus
         _assert_customer_model_limit_error(_chat(gateway, key_two, capped, acme), acme, capped)
         assert _chat(gateway, key_one, capped, globex).status_code == 200
         assert _chat(gateway, key_one, open_model, acme).status_code == 200
+
+
+@pytest.mark.parametrize(
+    ("limit_field", "limited_by"),
+    (("model_rpm_limit", "model_per_end_user"), ("rpm_limit", "end_user")),
+    ids=("customer-model", "customer-wide"),
+)
+def test_customer_rate_limit_429_names_the_limit_in_a_header(gateway: Gateway, limit_field: str, limited_by: str) -> None:
+    with gateway.scenario() as scenario:
+        capped: Final = scenario.model()
+        customer: Final = _customer(scenario, **{limit_field: {capped: 1} if limit_field.startswith("model_") else 1})
+        key: Final = scenario.key()
+
+        assert _chat(gateway, key, capped, customer).status_code == 200
+        rejected: Final = _chat(gateway, key, capped, customer)
+        assert rejected.status_code == 429, rejected.text
+        assert rejected.headers.get("rate_limit_type") == "requests", dict(rejected.headers)
+        assert rejected.headers.get("rate_limited_by") == limited_by, dict(rejected.headers)

tests/unit/proxy/hooks/test_parallel_request_limiter_v3.py

@@ -4688,6 +4688,7 @@ async def test_mcp_per_key_rejection_does_not_consume_shared_server_rpm_v3() ->
     assert key_rejected.value.headers is not None
     assert key_rejected.value.headers["retry-after"] == str(handler.window_size)
     assert key_rejected.value.headers["rate_limit_type"] == "requests"
+    assert key_rejected.value.headers["rate_limited_by"] == "mcp_per_key"
     assert key_rejected.value.headers["reset_at"]
     await handler.enforce_mcp_server_rate_limits(second_key, server)
 
@@ -8606,6 +8607,7 @@ def test_rate_limit_error_reports_reset_time_in_utc_on_a_non_utc_proxy(process_t
     assert exc_info.value.headers == {
         "retry-after": str(handler.window_size),
         "rate_limit_type": "requests",
+        "rate_limited_by": "api_key",
         "reset_at": expected_reset,
     }
     assert exc_info.value.detail == (
@@ -9938,6 +9940,8 @@ async def test_end_user_model_rpm_limit_is_shared_across_keys_and_isolated_per_c
         await _rpm_request(handler, cache, acme_via_key_two, "target")
     assert exc.value.status_code == 429
     assert "model_per_end_user: acme:target" in str(exc.value.detail), exc.value.detail
+    assert exc.value.headers is not None
+    assert exc.value.headers["rate_limited_by"] == "model_per_end_user"
 
     await _rpm_request(handler, cache, globex, "target")
     await _rpm_request(handler, cache, acme_via_key_one, "other-model")