fix(model_armor): gate apply_guardrail raise to non-logging_only and require native guardrails to declare logging_only

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

litellm/integrations/custom_guardrail.py

@@ -603,7 +603,9 @@ class CustomGuardrail(CustomLogger):
         supported_event_hooks: list[GuardrailEventHooks],
     ) -> None:
         allowed_hooks: Final = frozenset(supported_event_hooks) | (
-            frozenset((GuardrailEventHooks.logging_only,)) if self.uses_apply_guardrail_interface() else frozenset()
+            frozenset((GuardrailEventHooks.logging_only,))
+            if self.uses_apply_guardrail_interface() and not self.use_native_lifecycle_hooks
+            else frozenset()
         )
 
         def _validate_event_hook_list_is_in_supported_event_hooks(

litellm/proxy/guardrails/guardrail_hooks/model_armor/model_armor.py

@@ -1,7 +1,7 @@
 import time
 from collections.abc import AsyncGenerator, Mapping, Sequence
 from enum import Enum, auto
-from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, Optional
+from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal
 
 import httpx
 from fastapi import HTTPException
@@ -1232,7 +1232,7 @@ class ModelArmorGuardrail(CustomGuardrail, VertexBase):
         inputs: GenericGuardrailAPIInputs,
         request_data: dict,
         input_type: Literal["request", "response"],
-        logging_obj: Optional["LiteLLMLoggingObj"] = None,
+        logging_obj: "LiteLLMLoggingObj | None" = None,
     ) -> GenericGuardrailAPIInputs:
         content: Final = "\n".join(text for text in inputs.get("texts") or () if text)
         if not content:
@@ -1270,6 +1270,14 @@ class ModelArmorGuardrail(CustomGuardrail, VertexBase):
             end_time=end_time,
             duration=end_time - start_time,
         )
+        if flagged and not self._event_hook_is_event_type(GuardrailEventHooks.logging_only):
+            raise HTTPException(
+                status_code=400,
+                detail=self._build_block_error_detail(
+                    "Response blocked by Model Armor" if input_type == "response" else "Content blocked by Model Armor",
+                    armor_response,
+                ),
+            )
         return inputs
 
     @staticmethod

tests/test_litellm/integrations/test_custom_guardrail.py

@@ -2404,7 +2404,7 @@ def test_logging_only_requires_framework_support_or_explicit_declaration(
     event_hook: GuardrailEventHooks | str | list[GuardrailEventHooks] | list[str] | Mode,
 ) -> None:
     supported: Final = [GuardrailEventHooks.pre_call]
-    if guardrail_type is not CustomGuardrail:
+    if guardrail_type is _InheritedApplyGuardrail:
         guardrail: Final = guardrail_type(event_hook=event_hook, supported_event_hooks=supported)
         assert guardrail.event_hook == event_hook
         assert supported == [GuardrailEventHooks.pre_call]

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_model_armor.py

@@ -5255,3 +5255,30 @@ async def test_logging_only_flagged_prompt_still_scans_response():
     entries = _metadata_entries(out_kwargs)
     flagged = [e for e in entries if e["guardrail_status"] == "guardrail_flagged"]
     assert len(flagged) == 2
+
+
+@pytest.mark.asyncio
+async def test_apply_guardrail_raises_on_flagged_when_not_logging_only():
+    """The /guardrails/apply_guardrail endpoint calls apply_guardrail directly; a
+    non-logging_only instance must signal the block so flagged text is not returned as clean."""
+    guardrail = ModelArmorGuardrail(
+        template_id="test-template",
+        project_id="test-project",
+        location="us-central1",
+        guardrail_name="model-armor-pre",
+        event_hook=GuardrailEventHooks.pre_call,
+    )
+    guardrail.make_model_armor_request = AsyncMock(return_value=_flagged_armor_response())
+    request_data = {"metadata": {}}
+
+    with pytest.raises(HTTPException) as exc_info:
+        await guardrail.apply_guardrail(
+            inputs={"texts": ["forbidden prompt"]},
+            request_data=request_data,
+            input_type="request",
+        )
+
+    assert exc_info.value.status_code == 400
+    entries = request_data["metadata"]["standard_logging_guardrail_information"]
+    flagged = [e for e in entries if e["guardrail_status"] == "guardrail_flagged"]
+    assert len(flagged) == 1