fix(guardrails): classify all 4xx HTTPException guardrail blocks as intervened (#33821)

* fix(guardrails): classify all 4xx HTTPException guardrail blocks as intervened

* fix(guardrails): narrow HTTPException block classification to 400/403/422

---------

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

litellm/integrations/custom_guardrail.py

@@ -69,6 +69,8 @@ from litellm.exceptions import (
 # proxy's metadata sanitizer.
 _PRE_CALL_EXECUTED_TOKEN = secrets.token_hex(16)
 
+_GUARDRAIL_BLOCK_STATUS_CODES = frozenset({400, 403, 422})
+
 _guardrail_self_recorded: contextvars.ContextVar[bool] = contextvars.ContextVar(
     "litellm_guardrail_self_recorded", default=False
 )
@@ -1055,8 +1057,15 @@ class CustomGuardrail(CustomLogger):
         - GuardrailRaisedException (generic guardrail API, tool permission)
         - BlockedPiiEntityError (Presidio PII detection)
         - SensitiveDataRouteException (sensitive-data reroute to on-premise model)
-        - HTTPException with status 400 (content policy violation)
+        - HTTPException with a block-signalling status (400, 403, 422)
         - ModifyResponseException (passthrough mode violation)
+
+        Only the statuses guardrails use in-tree to signal a deliberate rejection
+        count as an intervention: 400 (content policy), 403 (e.g. akto) and 422
+        (e.g. llm_as_a_judge). Other 4xx codes are commonly propagated from an
+        upstream guardrail provider response (401 bad key, 408 timeout, 429 rate
+        limit, or a raw upstream status), which are technical failures, not
+        blocks, so they stay guardrail_failed_to_respond.
         """
         if isinstance(e, ModifyResponseException):
             return True
@@ -1069,7 +1078,11 @@ class CustomGuardrail(CustomLogger):
             ),
         ):
             return True
-        if HTTPException is not None and isinstance(e, HTTPException) and e.status_code == 400:
+        if (
+            HTTPException is not None
+            and isinstance(e, HTTPException)
+            and e.status_code in _GUARDRAIL_BLOCK_STATUS_CODES
+        ):
             return True
         return False
 

tests/test_litellm/integrations/test_custom_guardrail.py

@@ -1668,6 +1668,47 @@ class TestGuardrailInterventionClassification:
         )
         assert CustomGuardrail._is_guardrail_intervention(exc) is True
 
+    @pytest.mark.parametrize("status_code", [400, 403, 422])
+    def test_block_signalling_http_exception_is_intervention(self, status_code):
+        from fastapi.exceptions import HTTPException
+
+        exc = HTTPException(status_code=status_code, detail="blocked by guardrail")
+        assert CustomGuardrail._is_guardrail_intervention(exc) is True
+
+    @pytest.mark.parametrize("status_code", [300, 401, 408, 429, 451, 499, 500, 502, 503])
+    def test_non_block_http_exception_is_not_intervention(self, status_code):
+        from fastapi.exceptions import HTTPException
+
+        exc = HTTPException(status_code=status_code, detail="guardrail api error")
+        assert CustomGuardrail._is_guardrail_intervention(exc) is False
+
+    @pytest.mark.asyncio
+    async def test_non_400_4xx_logged_as_intervened_not_failed(self):
+        from fastapi.exceptions import HTTPException
+
+        from litellm.integrations.custom_guardrail import log_guardrail_information
+        from litellm.types.guardrails import GuardrailEventHooks
+
+        class BlockingGuardrail(CustomGuardrail):
+            def __init__(self):
+                super().__init__(
+                    guardrail_name="block-rail",
+                    event_hook=GuardrailEventHooks.pre_call,
+                )
+
+            @log_guardrail_information
+            async def async_pre_call_hook(self, data, **kwargs):
+                raise HTTPException(status_code=403, detail="blocked by guardrail")
+
+        guardrail = BlockingGuardrail()
+        request_data: dict = {"metadata": {}}
+
+        with pytest.raises(HTTPException):
+            await guardrail.async_pre_call_hook(data=request_data)
+
+        slg = request_data["metadata"]["standard_logging_guardrail_information"][0]
+        assert slg["guardrail_status"] == "guardrail_intervened"
+
     @pytest.mark.asyncio
     async def test_routing_logged_as_intervened_not_failed(self):
         from litellm.exceptions import SensitiveDataRouteException