Merge pull request #41583 from BerriAI/litellm_applied_guardrails_blocker

* fix(proxy): name the blocking guardrail in x-litellm-applied-guardrails

When a guardrail hook raises, the common ProxyLogging dispatch (sequential and parallel pre_call, pipeline block, during_call and post_call metrics wrapper, streaming iterator wrapper) now records that guardrail in applied_guardrails before re-raising, and pre_call_hook folds request-declared guardrails in on its raising path. Buffered streams rebuild their response headers after the first chunk so a post_call block reached while buffering carries the blocker too

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

* fix(proxy): attribute only the raising layer in stream and pipeline blocks

The streaming wrapper caught every exception crossing its boundary and named its own
callback, so a block by an inner guardrail or a provider stream failure also named every
outer guardrail. The wrapper now runs the hook over an upstream boundary that remembers
the exception it raised, and skips attribution when the same exception passes through

Pipeline blocks converted from SensitiveDataRouteException or ModifyResponseException into
a generic guardrail_pipeline_error now still record the blocking step's guardrail

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

* refactor(proxy): drop explanatory docstrings from the stream attribution helpers

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

---------

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

litellm/proxy/common_request_processing.py

@@ -2593,10 +2593,12 @@ class ProxyBaseLLMRequestProcessing:
 
                 async def refresh_stream_headers() -> Mapping[str, str]:
                     """`custom_headers` rebuilt for whichever deployment served the stream."""
-                    if not getattr(response, "fallback_headers_adopted", False):
-                        return custom_headers
                     return self._stream_response_headers(
-                        hidden_params=get_hidden_params_dict(response),
+                        hidden_params=(
+                            get_hidden_params_dict(response)
+                            if getattr(response, "fallback_headers_adopted", False)
+                            else hidden_params
+                        ),
                         user_api_key_dict=user_api_key_dict,
                         logging_obj=logging_obj,
                         version=version,

litellm/proxy/utils.py

@@ -12,13 +12,36 @@ import sys
 import threading
 import time
 import traceback
-from collections.abc import AsyncGenerator, Awaitable, Callable, Coroutine, Mapping, Sequence
+from collections.abc import (
+    AsyncGenerator,
+    AsyncIterable,
+    AsyncIterator,
+    Awaitable,
+    Callable,
+    Coroutine,
+    Mapping,
+    Sequence,
+)
 from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
+from functools import partial
 from types import MappingProxyType
-from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, Optional, Protocol, TypeVar, Union, cast, overload
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    ClassVar,
+    Final,
+    Generic,
+    Literal,
+    Optional,
+    Protocol,
+    TypeVar,
+    Union,
+    cast,
+    overload,
+)
 
 from typing_extensions import ReadOnly, TypedDict
 
@@ -123,6 +146,7 @@ from litellm.proxy._types import (
     UserAPIKeyAuth,
 )
 from litellm.proxy.auth.route_checks import RouteChecks
+from litellm.proxy.common_utils.callback_utils import add_guardrail_to_applied_guardrails_header
 from litellm.proxy.common_utils.config_sync_pubsub import publish_config_param_change
 from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
 from litellm.proxy.db.create_views import (
@@ -438,6 +462,36 @@ def _enrich_http_exception_with_guardrail_context(exc: BaseException, callback:
         detail.setdefault("guardrail_mode", event_hook)
 
 
+def _record_raising_guardrail(request_data: Mapping[str, object], callback: object) -> None:
+    guardrail_name: Final[object] = getattr(callback, "guardrail_name", None)
+    if isinstance(request_data, dict) and isinstance(guardrail_name, str):
+        add_guardrail_to_applied_guardrails_header(request_data=request_data, guardrail_name=guardrail_name)
+
+
+class _UpstreamStreamBoundary(Generic[_T]):
+    __slots__ = ("_upstream", "failure")
+
+    def __init__(self, upstream: AsyncIterable[_T]) -> None:
+        self._upstream: Final = upstream.__aiter__()
+        self.failure: BaseException | None = None
+
+    def __aiter__(self) -> "_UpstreamStreamBoundary[_T]":
+        return self
+
+    async def __anext__(self) -> _T:
+        try:
+            return await self._upstream.__anext__()
+        except StopAsyncIteration:
+            raise
+        except Exception as e:
+            self.failure = e
+            raise
+
+
+class _StreamIteratorHook(Protocol[_T]):
+    def __call__(self, *, response: AsyncIterator[_T]) -> AsyncGenerator[_T, None]: ...
+
+
 def _is_client_error_exception(exc: Exception) -> bool:
     if isinstance(exc, HTTPException):
         return exc.status_code < 500
@@ -1816,13 +1870,19 @@ class ProxyLogging:
         )
         if expected_if_unmutated is not None:
             callback.mark_pre_call_hook_ran(expected_if_unmutated)
-        result: Final = await self._process_guardrail_callback(
-            callback=callback,
-            data=input_data,
-            user_api_key_dict=user_api_key_dict,
-            call_type=call_type,
-            event_type=GuardrailEventHooks.pre_call,
-        )
+        try:
+            result: Final = await self._process_guardrail_callback(
+                callback=callback,
+                data=input_data,
+                user_api_key_dict=user_api_key_dict,
+                call_type=call_type,
+                event_type=GuardrailEventHooks.pre_call,
+            )
+        except SensitiveDataRouteException:
+            raise
+        except Exception:
+            _record_raising_guardrail(data, callback)
+            raise
         if (
             scans_raw_request
             and expected_if_unmutated is not None
@@ -2045,13 +2105,18 @@ class ProxyLogging:
             _merge_pipeline_metadata_writes(data, result.modified_data)
 
         if result.terminal_action == "block":
+            blocking_step: Final = result.step_results[-1] if result.step_results else None
+            callback: Final = (
+                PipelineExecutor.find_guardrail_callback(blocking_step.guardrail_name)
+                if blocking_step is not None
+                else None
+            )
+            if callback is not None:
+                _record_raising_guardrail(data, callback)
             original_exception: Final = result.original_exception
             if original_exception is not None and not _exception_changes_request_flow(original_exception):
-                blocking_step: Final = result.step_results[-1] if result.step_results else None
-                if blocking_step is not None:
-                    callback: Final = PipelineExecutor.find_guardrail_callback(blocking_step.guardrail_name)
-                    if callback is not None:
-                        _enrich_http_exception_with_guardrail_context(original_exception, callback)
+                if callback is not None:
+                    _enrich_http_exception_with_guardrail_context(original_exception, callback)
                 raise original_exception
 
             step_results_serializable: Final = [
@@ -2317,8 +2382,10 @@ class ProxyLogging:
             if data is not None:
                 self._process_guardrail_metadata(data)
             return data
-        except Exception as e:
-            raise e
+        except Exception:
+            if data is not None:
+                self._process_guardrail_metadata(data)
+            raise
 
     async def _run_parallel_pre_call_guardrails(
         self,
@@ -2376,6 +2443,8 @@ class ProxyLogging:
             # live kwargs.
             if callback.scan_raw_request and not isinstance(result, BaseException) and result is not None:
                 callback.mark_pre_call_hook_ran(data)
+            if isinstance(result, BaseException) and not isinstance(result, SensitiveDataRouteException):
+                _record_raising_guardrail(data, callback)
         raised: Final = tuple(result for result in results if isinstance(result, BaseException))
         blocking: Final = next((exc for exc in raised if not _exception_changes_request_flow(exc)), None)
         if blocking is not None:
@@ -2454,7 +2523,12 @@ class ProxyLogging:
                 break
 
     @staticmethod
-    async def _run_guardrail_with_metrics(callback: object, coro: Awaitable[_T], hook_type: str) -> _T:
+    async def _run_guardrail_with_metrics(
+        callback: object,
+        coro: Awaitable[_T],
+        hook_type: str,
+        request_data: Mapping[str, object],
+    ) -> _T:
         """
         Await `coro`, recording its latency and status to the
         `litellm_guardrail_latency_seconds` metric under `hook_type`, and
@@ -2474,6 +2548,7 @@ class ProxyLogging:
             status = "error"
             error_type = type(e).__name__
             _enrich_http_exception_with_guardrail_context(e, callback)
+            _record_raising_guardrail(request_data, callback)
             raise
         finally:
             ProxyLogging._emit_guardrail_metrics(
@@ -2486,21 +2561,19 @@ class ProxyLogging:
 
     @staticmethod
     async def _wrap_streaming_iterator_with_enrichment(
-        callback: object, gen: AsyncGenerator[_T, None]
+        callback: object,
+        response: AsyncIterable[_T],
+        hook: _StreamIteratorHook[_T],
+        request_data: Mapping[str, object],
     ) -> AsyncGenerator[_T, None]:
-        """
-        Yield from `gen`; if iteration raises an HTTPException with dict detail,
-        enrich the detail with the originating callback's `guardrail_name` and
-        `guardrail_mode` before re-raising. Used to wrap each layer of the
-        async_post_call_streaming_iterator_hook chain so the enrichment is
-        attributed to the callback that produced the chunk pipeline at that
-        point in the chain.
-        """
+        upstream: Final = _UpstreamStreamBoundary(response)
         try:
-            async for chunk in gen:
+            async for chunk in hook(response=upstream):
                 yield chunk
         except Exception as e:
-            _enrich_http_exception_with_guardrail_context(e, callback)
+            if e is not upstream.failure:
+                _enrich_http_exception_with_guardrail_context(e, callback)
+                _record_raising_guardrail(request_data, callback)
             raise
 
     # Cache for callback-capability detection. Keyed on a signature of
@@ -2735,6 +2808,7 @@ class ProxyLogging:
                     call_type=call_type,
                 ),
                 "during_call",
+                request_data=data,
             )
             return
         await self._run_guardrail_with_metrics(
@@ -2745,6 +2819,7 @@ class ProxyLogging:
                 call_type=call_type,
             ),
             "during_call",
+            request_data=data,
         )
 
     async def failed_tracking_alert(
@@ -3263,6 +3338,7 @@ class ProxyLogging:
                             response=response,
                         ),
                         "post_call",
+                        request_data=data,
                     )
                 else:
                     guardrail_response = await self._run_guardrail_with_metrics(
@@ -3273,6 +3349,7 @@ class ProxyLogging:
                             response=response,
                         ),
                         "post_call",
+                        request_data=data,
                     )
 
                 if guardrail_response is not None:
@@ -3336,6 +3413,7 @@ class ProxyLogging:
                         response=response,
                     ),
                     "post_call",
+                    request_data=data,
                 )
             else:
                 await self._run_guardrail_with_metrics(
@@ -3346,6 +3424,7 @@ class ProxyLogging:
                         response=response,
                     ),
                     "post_call",
+                    request_data=data,
                 )
 
         results: Final = await asyncio.gather(
@@ -3409,6 +3488,7 @@ class ProxyLogging:
                     request_data=request_data,
                 ),
                 "post_mcp_call",
+                request_data=request_data,
             )
         return response
 
@@ -3650,27 +3730,27 @@ class ProxyLogging:
                 )
                 else kind
             )
-            if effective_kind == "override":
-                current_response = self._wrap_streaming_iterator_with_enrichment(
-                    resolved_callback,
-                    resolved_callback.async_post_call_streaming_iterator_hook(
-                        user_api_key_dict=user_api_key_dict,
-                        response=current_response,
-                        request_data=request_data,
-                    ),
+            hook: _StreamIteratorHook[object] = (
+                partial(
+                    resolved_callback.async_post_call_streaming_iterator_hook,
+                    user_api_key_dict=user_api_key_dict,
+                    request_data=request_data,
                 )
-            else:
-                # kind == "apply_guardrail": route through unified_guardrail
-                current_response = self._wrap_streaming_iterator_with_enrichment(
-                    resolved_callback,
-                    unified_guardrail.async_post_call_streaming_iterator_hook(
-                        user_api_key_dict=user_api_key_dict,
-                        request_data=request_data,
-                        response=current_response,
-                        guardrail_to_apply=resolved_callback,
-                        buffer_until_moderated_default=(kind == "override"),
-                    ),
+                if effective_kind == "override"
+                else partial(
+                    unified_guardrail.async_post_call_streaming_iterator_hook,
+                    user_api_key_dict=user_api_key_dict,
+                    request_data=request_data,
+                    guardrail_to_apply=resolved_callback,
+                    buffer_until_moderated_default=(kind == "override"),
                 )
+            )
+            current_response = self._wrap_streaming_iterator_with_enrichment(
+                resolved_callback,
+                current_response,
+                hook,
+                request_data=request_data,
+            )
 
         pipeline_translation: Final = (
             resolve_endpoint_translation(user_api_key_dict, None) if post_call_pipelines else None

tests/test_litellm/proxy/test_common_request_processing.py

@@ -40,9 +40,12 @@ from litellm.proxy.common_request_processing import (
     _parse_event_data_for_error,
     _resolve_per_request_model_group_alias,
     _should_return_raw_model_name,
+    _sse_error_frames,
     _UpstreamClosingStreamingResponse,
     create_response,
+    sse_error_payload,
 )
+from litellm.proxy.common_utils.callback_utils import add_guardrail_to_applied_guardrails_header
 from litellm.proxy.dd_span_tagger import DDSpanTagger
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.proxy._types import ProxyErrorTypes, ProxyException
@@ -9065,6 +9068,60 @@ class TestStreamingResponseHeadersFollowFallback:
         assert "llm_provider-stale-marker" not in result.headers
         assert result.headers["x-callback-header"] == "kept"
 
+    @pytest.mark.asyncio
+    async def test_streaming_block_headers_name_the_blocking_guardrail(self, monkeypatch):
+        processor_data: dict[str, object] = {"model": "oa", "stream": True, "metadata": {}}
+
+        def select_data_generator(**kwargs):
+            async def generator():
+                add_guardrail_to_applied_guardrails_header(processor_data, "stream-blocker")
+                _, error_obj = sse_error_payload(HTTPException(status_code=400, detail="blocked"))
+                for frame in _sse_error_frames(error_obj):
+                    yield frame
+
+            return generator()
+
+        logging_obj = MagicMock()
+        logging_obj.litellm_call_id = "lit-7144-call"
+        logging_obj._defer_async_logging = False
+        logging_obj._on_deferred_stream_complete = None
+        logging_obj.cost_breakdown = None
+        processor_data["litellm_logging_obj"] = logging_obj
+        processor = ProxyBaseLLMRequestProcessing(data=processor_data)
+
+        proxy_logging_obj = MagicMock(spec=ProxyLogging)
+        proxy_logging_obj.during_call_hook = AsyncMock(return_value=None)
+        proxy_logging_obj.update_request_status = AsyncMock(return_value=None)
+        proxy_logging_obj.post_call_success_hook = AsyncMock(
+            side_effect=lambda data, user_api_key_dict, response: response
+        )
+        proxy_logging_obj.post_call_response_headers_hook = AsyncMock(return_value={})
+
+        async def fake_route_request(**kwargs):
+            async def call():
+                return SimpleNamespace(_hidden_params={}, fallback_headers_adopted=False)
+
+            return call()
+
+        monkeypatch.setattr(litellm.proxy.common_request_processing, "route_request", fake_route_request)
+
+        result = await processor.base_process_llm_request(
+            request=Request(scope={"type": "http", "headers": []}),
+            fastapi_response=Response(),
+            user_api_key_dict=ProxyUserAPIKeyAuth(api_key="sk-test"),
+            route_type="acompletion",
+            proxy_logging_obj=proxy_logging_obj,
+            general_settings={},
+            proxy_config=MagicMock(spec=ProxyConfig),
+            select_data_generator=select_data_generator,
+            is_streaming_request=True,
+            skip_pre_call_logic=True,
+        )
+
+        assert isinstance(result, JSONResponse)
+        assert result.status_code == 400
+        assert result.headers["x-litellm-applied-guardrails"] == "stream-blocker"
+
 
 class _MessagesFallbackStream:
     def __init__(self) -> None:

tests/test_litellm/proxy/utils/proxy_logging/test_during_call_hook.py

@@ -6,6 +6,7 @@ from typing import Any, Dict
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from fastapi import HTTPException
 
 import litellm
 from litellm.integrations.custom_guardrail import CustomGuardrail
@@ -84,3 +85,20 @@ async def test_during_call_hook_guardrail_error_raises(proxy_logging, make_user_
             user_api_key_dict=make_user_api_key_auth(),
             call_type="completion",
         )
+
+
+@pytest.mark.asyncio
+async def test_during_call_block_names_the_blocking_guardrail_in_applied_guardrails(
+    proxy_logging, make_user_api_key_auth, monkeypatch
+):
+    g = _make_guardrail("blocker")
+    g.async_moderation_hook = AsyncMock(side_effect=HTTPException(status_code=400, detail="blocked"))
+    monkeypatch.setattr(litellm, "callbacks", [_make_guardrail("passer"), g])
+    data = {"model": "m", "metadata": {}}
+    with pytest.raises(HTTPException):
+        await proxy_logging.during_call_hook(
+            data=data,
+            user_api_key_dict=make_user_api_key_auth(),
+            call_type="completion",
+        )
+    assert "blocker" in data["metadata"]["applied_guardrails"]

tests/test_litellm/proxy/utils/proxy_logging/test_guardrail_pipeline.py

@@ -527,17 +527,19 @@ def test_handle_pipeline_result_block_enriches_with_guardrail_name_and_mode():
     result.step_results = [MagicMock(guardrail_name="g")]
     result.original_exception = original
 
+    data: dict[str, object] = {"model": "m"}
     saved = litellm.callbacks
     litellm.callbacks = [cb]
     try:
         with pytest.raises(HTTPException) as info:
-            ProxyLogging._handle_pipeline_result(result=result, data={"model": "m"}, policy_name="p")
+            ProxyLogging._handle_pipeline_result(result=result, data=data, policy_name="p")
     finally:
         litellm.callbacks = saved
 
     assert info.value is original
     assert info.value.detail["guardrail_name"] == "g"
     assert info.value.detail["guardrail_mode"] == GuardrailEventHooks.pre_call
+    assert data["metadata"] == {"applied_guardrails": ["g"]}
 
 
 def test_handle_pipeline_result_block_does_not_reraise_sensitive_data_route():
@@ -549,14 +551,23 @@ def test_handle_pipeline_result_block_does_not_reraise_sensitive_data_route():
         session_id="sess-1",
         guardrail_name="pii-router",
     )
+    cb = _make_guardrail()
+    cb.guardrail_name = "pii-router"
     result = MagicMock()
     result.terminal_action = "block"
     result.step_results = [MagicMock(guardrail_name="pii-router")]
     result.original_exception = original
-    with pytest.raises(HTTPException) as info:
-        ProxyLogging._handle_pipeline_result(result=result, data={"model": "m"}, policy_name="p")
+    data: dict[str, object] = {"model": "m"}
+    saved = litellm.callbacks
+    litellm.callbacks = [cb]
+    try:
+        with pytest.raises(HTTPException) as info:
+            ProxyLogging._handle_pipeline_result(result=result, data=data, policy_name="p")
+    finally:
+        litellm.callbacks = saved
     assert info.value.status_code == 400
     assert info.value.detail["error"]["type"] == "guardrail_pipeline_error"
+    assert data["metadata"] == {"applied_guardrails": ["pii-router"]}
 
 
 def test_handle_pipeline_result_block_does_not_reraise_modify_response():
@@ -569,14 +580,23 @@ def test_handle_pipeline_result_block_does_not_reraise_modify_response():
         request_data={"model": "m"},
         guardrail_name="masker",
     )
+    cb = _make_guardrail()
+    cb.guardrail_name = "masker"
     result = MagicMock()
     result.terminal_action = "block"
     result.step_results = [MagicMock(guardrail_name="masker")]
     result.original_exception = original
-    with pytest.raises(HTTPException) as info:
-        ProxyLogging._handle_pipeline_result(result=result, data={"model": "m"}, policy_name="p")
+    data: dict[str, object] = {"model": "m"}
+    saved = litellm.callbacks
+    litellm.callbacks = [cb]
+    try:
+        with pytest.raises(HTTPException) as info:
+            ProxyLogging._handle_pipeline_result(result=result, data=data, policy_name="p")
+    finally:
+        litellm.callbacks = saved
     assert info.value.status_code == 400
     assert info.value.detail["error"]["type"] == "guardrail_pipeline_error"
+    assert data["metadata"] == {"applied_guardrails": ["masker"]}
 
 
 def test_handle_pipeline_result_modify_response_raises_modify_exception():
@@ -617,7 +637,7 @@ async def test_run_guardrail_with_metrics_passes_result_and_records_success(monk
     monkeypatch.setattr(litellm, "callbacks", [prom])
 
     out = await ProxyLogging._run_guardrail_with_metrics(
-        callback=MagicMock(guardrail_name="g"), coro=task(), hook_type="during_call"
+        callback=MagicMock(guardrail_name="g"), coro=task(), hook_type="during_call", request_data={}
     )
 
     assert out == {"a": 1, "b": 2, "c": 3}
@@ -643,7 +663,7 @@ async def test_run_guardrail_with_metrics_records_error_and_enriches(monkeypatch
     monkeypatch.setattr(litellm, "callbacks", [prom])
 
     with pytest.raises(HTTPException):
-        await ProxyLogging._run_guardrail_with_metrics(callback=cb, coro=task(), hook_type="post_call")
+        await ProxyLogging._run_guardrail_with_metrics(callback=cb, coro=task(), hook_type="post_call", request_data={})
 
     assert detail["guardrail_name"] == "presidio"
     recorded = prom._record_guardrail_metrics.call_args.kwargs

tests/test_litellm/proxy/utils/proxy_logging/test_post_call_success_hook.py

@@ -6,6 +6,7 @@ from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from fastapi import HTTPException
 
 import litellm
 from litellm.integrations.custom_guardrail import CustomGuardrail
@@ -96,3 +97,26 @@ async def test_post_call_success_hook_guardrail_returns_modified_response(
         data={}, response={"orig": True}, user_api_key_dict=make_user_api_key_auth()
     )
     assert out == modified
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("run_in_parallel", [False, True], ids=["sequential", "parallel"])
+async def test_post_call_block_names_the_blocking_guardrail_in_applied_guardrails(
+    proxy_logging, make_user_api_key_auth, monkeypatch, run_in_parallel
+):
+    def _passer_that_records(data, user_api_key_dict, response):
+        data["metadata"]["applied_guardrails"] = ["passer"]
+
+    passer = _make_guardrail("passer")
+    passer.async_post_call_success_hook = AsyncMock(side_effect=_passer_that_records)
+    passer.run_in_parallel = run_in_parallel
+    blocker = _make_guardrail("blocker")
+    blocker.async_post_call_success_hook = AsyncMock(side_effect=HTTPException(status_code=400, detail="blocked"))
+    blocker.run_in_parallel = run_in_parallel
+    monkeypatch.setattr(litellm, "callbacks", [passer, blocker])
+    data = {"model": "m", "metadata": {}}
+    with pytest.raises(HTTPException):
+        await proxy_logging.post_call_success_hook(
+            data=data, response=MagicMock(), user_api_key_dict=make_user_api_key_auth()
+        )
+    assert data["metadata"]["applied_guardrails"] == ["passer", "blocker"]

tests/test_litellm/proxy/utils/proxy_logging/test_pre_call_hook.py

@@ -905,3 +905,43 @@ async def test_scan_raw_request_warns_on_in_place_mutation_returning_none(
     )
     mock_logger.warning.assert_called_once()
     assert "scan_raw_request" in str(mock_logger.warning.call_args)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "blocker_kwargs",
+    [
+        pytest.param({}, id="sequential"),
+        pytest.param({"scan_raw_request": True}, id="scan_raw_request"),
+        pytest.param({"run_in_parallel": True}, id="parallel"),
+    ],
+)
+async def test_pre_call_block_names_the_blocking_guardrail_in_applied_guardrails(
+    proxy_logging, make_user_api_key_auth, monkeypatch, blocker_kwargs
+):
+    monkeypatch.setattr(litellm, "callbacks", [_BlockOnSecretGuardrail(**blocker_kwargs)])
+    proxy_logging.slack_alerting_instance = MagicMock(alerting=None)
+    data = _secret_request()
+    with pytest.raises(HTTPException, match="blocked"):
+        await proxy_logging.pre_call_hook(
+            user_api_key_dict=make_user_api_key_auth(),
+            data=data,
+            call_type="completion",
+        )
+    assert data["metadata"]["applied_guardrails"] == ["blocker"]
+
+
+@pytest.mark.asyncio
+async def test_pre_call_block_keeps_request_declared_guardrail_in_applied_guardrails(
+    proxy_logging, make_user_api_key_auth, monkeypatch
+):
+    monkeypatch.setattr(litellm, "callbacks", [_BlockOnSecretGuardrail(default_on=False)])
+    proxy_logging.slack_alerting_instance = MagicMock(alerting=None)
+    data = {**_secret_request(), "metadata": {"guardrails": ["blocker", "declared-post-call"]}}
+    with pytest.raises(HTTPException, match="blocked"):
+        await proxy_logging.pre_call_hook(
+            user_api_key_dict=make_user_api_key_auth(),
+            data=data,
+            call_type="completion",
+        )
+    assert data["metadata"]["applied_guardrails"] == ["blocker", "declared-post-call"]

tests/test_litellm/proxy/utils/proxy_logging/test_streaming_hooks.py

@@ -20,6 +20,7 @@ from fastapi import HTTPException
 
 import litellm
 from litellm.exceptions import GuardrailRaisedException
+from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
@@ -27,6 +28,7 @@ from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterato
 )
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.utils import ProxyLogging
+from litellm.types.guardrails import GuardrailEventHooks
 from litellm.types.utils import Usage
 
 
@@ -168,6 +170,15 @@ def test_init_response_taking_too_long_task_no_slack_instance_no_error_raises(pr
 # ---------------------------------------------------------------------------
 
 
+async def _passthrough_hook(*, response: AsyncIterator[object]) -> AsyncGenerator[object, None]:
+    async for chunk in response:
+        yield chunk
+
+
+async def _one_chunk() -> AsyncGenerator[object, None]:
+    yield "chunk"
+
+
 @pytest.mark.asyncio
 async def test_wrap_streaming_iterator_with_enrichment_passes_through_chunks(proxy_logging):
     async def gen():
@@ -175,7 +186,9 @@ async def test_wrap_streaming_iterator_with_enrichment_passes_through_chunks(pro
             yield ch
 
     cb = MagicMock(guardrail_name="g", event_hook="pre_call")
-    wrapped = proxy_logging._wrap_streaming_iterator_with_enrichment(callback=cb, gen=gen())
+    wrapped = proxy_logging._wrap_streaming_iterator_with_enrichment(
+        callback=cb, response=gen(), hook=_passthrough_hook, request_data={}
+    )
     out = [ch async for ch in wrapped]
     snapshot = {
         "chunks": out,
@@ -195,18 +208,43 @@ async def test_wrap_streaming_iterator_with_enrichment_passes_through_chunks(pro
 async def test_wrap_streaming_iterator_with_enrichment_enriches_http_exception_raises(proxy_logging):
     detail = {"error": "blocked"}
 
-    async def boom_gen():
+    async def boom_hook(*, response: AsyncIterator[object]) -> AsyncGenerator[object, None]:
         if False:
             yield  # pragma: no cover
         raise HTTPException(status_code=400, detail=detail)
 
     cb = MagicMock(guardrail_name="presidio", event_hook="post_call")
-    wrapped = proxy_logging._wrap_streaming_iterator_with_enrichment(callback=cb, gen=boom_gen())
+    request_data: dict[str, object] = {}
+    wrapped = proxy_logging._wrap_streaming_iterator_with_enrichment(
+        callback=cb, response=_one_chunk(), hook=boom_hook, request_data=request_data
+    )
     with pytest.raises(HTTPException):
         async for _ in wrapped:
             pass
     assert detail["guardrail_name"] == "presidio"
     assert detail["guardrail_mode"] == "post_call"
+    assert request_data["metadata"]["applied_guardrails"] == ["presidio"]
+
+
+@pytest.mark.asyncio
+async def test_wrap_streaming_iterator_leaves_upstream_http_exception_unattributed(proxy_logging):
+    detail = {"error": "upstream rejected the stream"}
+
+    async def failing_upstream() -> AsyncGenerator[object, None]:
+        if False:
+            yield  # pragma: no cover
+        raise HTTPException(status_code=502, detail=detail)
+
+    cb = MagicMock(guardrail_name="presidio", event_hook="post_call")
+    request_data: dict[str, object] = {}
+    wrapped = proxy_logging._wrap_streaming_iterator_with_enrichment(
+        callback=cb, response=failing_upstream(), hook=_passthrough_hook, request_data=request_data
+    )
+    with pytest.raises(HTTPException):
+        async for _ in wrapped:
+            pass
+    assert detail == {"error": "upstream rejected the stream"}
+    assert request_data == {}
 
 
 # ---------------------------------------------------------------------------
@@ -696,3 +734,85 @@ async def test_post_call_response_headers_hook_swallows_callback_error(proxy_log
         data={}, user_api_key_dict=make_user_api_key_auth(), response=response
     )
     assert out == {}
+
+
+class _StreamBlocker(CustomGuardrail):
+    def __init__(self, guardrail_name: str = "stream-blocker") -> None:
+        super().__init__(guardrail_name=guardrail_name, event_hook=GuardrailEventHooks.post_call, default_on=True)
+
+    async def async_post_call_streaming_iterator_hook(
+        self, user_api_key_dict: UserAPIKeyAuth, response: AsyncIterator[object], request_data: dict[str, object]
+    ) -> AsyncGenerator[object, None]:
+        async for _ in response:
+            raise HTTPException(status_code=400, detail={"error": "blocked"})
+            yield  # pragma: no cover
+
+
+class _StreamPasser(CustomGuardrail):
+    def __init__(self, guardrail_name: str = "stream-passer") -> None:
+        super().__init__(guardrail_name=guardrail_name, event_hook=GuardrailEventHooks.post_call, default_on=True)
+
+    async def async_post_call_streaming_iterator_hook(
+        self, user_api_key_dict: UserAPIKeyAuth, response: AsyncIterator[object], request_data: dict[str, object]
+    ) -> AsyncGenerator[object, None]:
+        async for chunk in response:
+            yield chunk
+
+
+async def _drain_stream_chain(
+    proxy_logging: ProxyLogging,
+    user_api_key_dict: UserAPIKeyAuth,
+    upstream: AsyncIterator[object],
+    request_data: dict[str, object],
+) -> None:
+    async for _ in proxy_logging.async_post_call_streaming_iterator_hook(
+        response=upstream,
+        user_api_key_dict=user_api_key_dict,
+        request_data=request_data,
+    ):
+        pass
+
+
+async def _failing_provider_stream() -> AsyncGenerator[object, None]:
+    yield "chunk"
+    raise RuntimeError("provider connection dropped")
+
+
+@pytest.mark.asyncio
+async def test_stream_guardrail_block_names_the_blocking_guardrail_in_applied_guardrails(
+    proxy_logging, make_user_api_key_auth, monkeypatch
+):
+    monkeypatch.setattr(litellm, "callbacks", [_StreamBlocker()])
+    monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None, raising=False)
+
+    request_data: dict[str, object] = {"metadata": {}}
+    with pytest.raises(HTTPException):
+        await _drain_stream_chain(proxy_logging, make_user_api_key_auth(), _one_chunk(), request_data)
+    assert request_data["metadata"]["applied_guardrails"] == ["stream-blocker"]
+
+
+@pytest.mark.asyncio
+async def test_stream_block_by_inner_guardrail_does_not_name_the_outer_layers(
+    proxy_logging, make_user_api_key_auth, monkeypatch
+):
+    monkeypatch.setattr(litellm, "callbacks", [_StreamBlocker(), _StreamPasser("outer-a"), _StreamPasser("outer-b")])
+    monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None, raising=False)
+
+    request_data: dict[str, object] = {"metadata": {}}
+    with pytest.raises(HTTPException) as info:
+        await _drain_stream_chain(proxy_logging, make_user_api_key_auth(), _one_chunk(), request_data)
+    assert info.value.detail["guardrail_name"] == "stream-blocker"
+    assert request_data["metadata"]["applied_guardrails"] == ["stream-blocker"]
+
+
+@pytest.mark.asyncio
+async def test_stream_provider_failure_is_not_attributed_to_any_guardrail(
+    proxy_logging, make_user_api_key_auth, monkeypatch
+):
+    monkeypatch.setattr(litellm, "callbacks", [_StreamPasser("outer-a"), _StreamPasser("outer-b")])
+    monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None, raising=False)
+
+    request_data: dict[str, object] = {"metadata": {}}
+    with pytest.raises(RuntimeError, match="provider connection dropped"):
+        await _drain_stream_chain(proxy_logging, make_user_api_key_auth(), _failing_provider_stream(), request_data)
+    assert request_data["metadata"] == {}