fix(guardrails): match policy-pipeline block response to direct guardrail attachment

litellm/proxy/policy_engine/pipeline_executor.py

@@ -6,7 +6,7 @@
 """
 
 import time
-from typing import Any, List, Optional
+from typing import Any, List, Literal, Optional
 
 import litellm
 from litellm._logging import verbose_proxy_logger
@@ -64,7 +64,12 @@ async def execute_steps(
         for i, step in enumerate(steps):
             start_time = time.perf_counter()
 
-            outcome, modified_data, error_detail = await PipelineExecutor._run_step(
+            (
+                outcome,
+                modified_data,
+                error_detail,
+                original_exception,
+            ) = await PipelineExecutor._run_step(
                 step=step,
                 mode=mode,
                 data=working_data,
@@ -108,6 +113,7 @@ async def execute_steps(
                     terminal_action="block",
                     step_results=step_results,
                     error_message=error_detail,
+                    original_exception=original_exception,
                 )
 
             if action == "modify_response":
@@ -134,22 +140,30 @@ async def _run_step(
         data: dict,
         user_api_key_dict: Any,
         call_type: str,
-    ) -> tuple:
+    ) -> tuple[
+        Literal["pass", "fail", "error"],
+        Optional[dict],
+        Optional[str],
+        Optional[Exception],
+    ]:
         """
         Run a single pipeline step's guardrail.
 
         Returns:
-            Tuple of (outcome, modified_data, error_detail) where:
+            Tuple of (outcome, modified_data, error_detail, original_exception):
             - outcome: "pass", "fail", or "error"
             - modified_data: dict if guardrail returned modified data, else None
             - error_detail: error message string if fail/error, else None
+            - original_exception: the exception the guardrail raised, so the
+              pipeline can re-raise it verbatim and match the direct-attachment
+              response/trace, else None
         """
-        callback = PipelineExecutor._find_guardrail_callback(step.guardrail)
+        callback = PipelineExecutor.find_guardrail_callback(step.guardrail)
         if callback is None:
             verbose_proxy_logger.warning(
                 f"Pipeline: guardrail '{step.guardrail}' not found in callbacks"
             )
-            return ("error", None, f"Guardrail '{step.guardrail}' not found")
+            return ("error", None, f"Guardrail '{step.guardrail}' not found", None)
 
         try:
             # Inject guardrail name into metadata so should_run_guardrail() allows it
@@ -182,26 +196,26 @@ async def _run_step(
                     response=data.get("response"),  # type: ignore
                 )
             else:
-                return ("error", None, f"Unsupported pipeline mode: {mode}")
+                return ("error", None, f"Unsupported pipeline mode: {mode}", None)
 
             # Normal return means pass
             modified_data = None
             if response is not None and isinstance(response, dict):
                 modified_data = response
-            return ("pass", modified_data, None)
+            return ("pass", modified_data, None, None)
 
         except Exception as e:
             if CustomGuardrail._is_guardrail_intervention(e):
                 error_msg = _extract_error_message(e)
-                return ("fail", None, error_msg)
+                return ("fail", None, error_msg, e)
             else:
                 verbose_proxy_logger.error(
                     f"Pipeline: unexpected error from guardrail '{step.guardrail}': {e}"
                 )
-                return ("error", None, str(e))
+                return ("error", None, str(e), e)
 
     @staticmethod
-    def _find_guardrail_callback(guardrail_name: str) -> Optional[CustomGuardrail]:
+    def find_guardrail_callback(guardrail_name: str) -> Optional[CustomGuardrail]:
         """Look up an initialized guardrail callback by name from litellm.callbacks."""
         for callback in litellm.callbacks:
             if isinstance(callback, CustomGuardrail):

litellm/proxy/utils.py

@@ -363,6 +363,16 @@ def _enrich_http_exception_with_guardrail_context(
         detail.setdefault("guardrail_mode", event_hook)
 
 
+def _exception_changes_request_flow(exc: BaseException) -> bool:
+    """
+    True for guardrail exceptions the proxy turns into an alternate request flow
+    (a reroute or a passthrough response) rather than a block. A pipeline step
+    configured to block must honor that block, so these are surfaced as the
+    generic pipeline block instead of being re-raised verbatim.
+    """
+    return isinstance(exc, (SensitiveDataRouteException, ModifyResponseException))
+
+
 @dataclass(frozen=True)
 class _CallbackCapabilities:
     """Cached per-hook capability flags derived from ``litellm.callbacks``.
@@ -1356,7 +1366,7 @@ async def _maybe_execute_pipelines(
 
     @staticmethod
     def _handle_pipeline_result(
-        result: Any,
+        result: PipelineExecutionResult,
         data: dict,
         policy_name: str,
     ) -> dict:
@@ -1371,6 +1381,21 @@ def _handle_pipeline_result(
             return data
 
         if result.terminal_action == "block":
+            original_exception = result.original_exception
+            if original_exception is not None and not _exception_changes_request_flow(
+                original_exception
+            ):
+                blocking_step = result.step_results[-1] if result.step_results else None
+                if blocking_step is not None:
+                    callback = PipelineExecutor.find_guardrail_callback(
+                        blocking_step.guardrail_name
+                    )
+                    if callback is not None:
+                        _enrich_http_exception_with_guardrail_context(
+                            original_exception, callback
+                        )
+                raise original_exception
+
             step_results_serializable = [
                 {
                     "guardrail": sr.guardrail_name,

litellm/types/proxy/policy_engine/pipeline_types.py

@@ -99,8 +99,11 @@ class PipelineStepResult(BaseModel):
 class PipelineExecutionResult(BaseModel):
     """Result of executing an entire pipeline."""
 
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
     terminal_action: str  # block | allow | modify_response
     step_results: List[PipelineStepResult]
     modified_data: Optional[Dict[str, Any]] = None
     error_message: Optional[str] = None
     modify_response_message: Optional[str] = None
+    original_exception: Optional[Exception] = Field(default=None, exclude=True)

tests/test_litellm/proxy/policy_engine/test_pipeline_executor.py

@@ -209,6 +209,75 @@ async def test_escalation_step1_fails_step2_blocks():
         litellm.callbacks = original_callbacks
 
 
+@pytest.mark.skipif(HTTPException is None, reason="fastapi not installed")
+@pytest.mark.asyncio
+async def test_block_carries_original_guardrail_exception():
+    """A blocking step must expose the guardrail's own raised exception on the
+    result so the caller can re-raise it verbatim, giving the policy path the
+    same response/trace as a direct guardrail attachment."""
+    guard = AlwaysFailGuardrail(guardrail_name="moderation-filter")
+
+    pipeline = GuardrailPipeline(
+        mode="pre_call",
+        steps=[
+            PipelineStep(
+                guardrail="moderation-filter", on_fail="block", on_pass="allow"
+            )
+        ],
+    )
+
+    original_callbacks = litellm.callbacks.copy()
+    litellm.callbacks = [guard]
+
+    try:
+        result = await PipelineExecutor.execute_steps(
+            steps=pipeline.steps,
+            mode=pipeline.mode,
+            data={"messages": [{"role": "user", "content": "bad content"}]},
+            user_api_key_dict=MagicMock(),
+            call_type="completion",
+            policy_name="content-safety",
+        )
+
+        assert result.terminal_action == "block"
+        assert isinstance(result.original_exception, HTTPException)
+        assert result.original_exception.status_code == 400
+        assert result.original_exception.detail == "Content policy violation"
+    finally:
+        litellm.callbacks = original_callbacks
+
+
+@pytest.mark.asyncio
+async def test_unsupported_mode_yields_error_outcome_without_exception():
+    """An unexpected hook mode must surface as an error outcome (carrying no
+    original exception), not crash or run the guardrail."""
+    guard = AlwaysPassGuardrail(guardrail_name="filter")
+
+    original_callbacks = litellm.callbacks.copy()
+    litellm.callbacks = [guard]
+
+    try:
+        result = await PipelineExecutor.execute_steps(
+            steps=[PipelineStep(guardrail="filter", on_error="block", on_fail="block")],
+            mode="during_call",
+            data={"messages": [{"role": "user", "content": "hi"}]},
+            user_api_key_dict=MagicMock(),
+            call_type="completion",
+            policy_name="content-safety",
+        )
+
+        assert guard.calls == 0
+        assert result.terminal_action == "block"
+        assert result.step_results[0].outcome == "error"
+        assert (
+            "Unsupported pipeline mode: during_call"
+            in result.step_results[0].error_detail
+        )
+        assert result.original_exception is None
+    finally:
+        litellm.callbacks = original_callbacks
+
+
 @pytest.mark.asyncio
 async def test_passthrough_guardrail_failure_can_pipeline_block():
     """