Merge pull request #41407 from BerriAI/litellm_content_filter_stream_bounded_scan

perf(content_filter): scan a bounded window per streamed chunk

litellm/constants.py

@@ -364,6 +364,8 @@ GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS: Final = int(
     os.getenv("GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS", 24 * 60 * 60)
 )
 BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS: Final = 25_000
+CONTENT_FILTER_STREAMING_HOLDBACK_CHARS: Final = 50
+CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS: Final = 512
 DEFAULT_PRESIDIO_ANALYZE_CHUNK_SIZE_BYTES: Final = 500_000
 PRESIDIO_ANALYZE_CHUNK_OVERLAP_CHARS: Final = 4096
 PRESIDIO_ANALYZE_CHUNK_CONCURRENCY: Final = 8

litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py

@@ -11,6 +11,7 @@ import os
 import re
 import time
 from collections.abc import AsyncGenerator, Coroutine, Mapping, Sequence
+from dataclasses import dataclass, replace
 from datetime import datetime
 from re import Pattern
 from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict, cast
@@ -20,7 +21,11 @@ from fastapi import HTTPException
 
 from litellm import Router
 from litellm._logging import verbose_proxy_logger
-from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
+from litellm.constants import (
+    CONTENT_FILTER_STREAMING_HOLDBACK_CHARS,
+    CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS,
+    DEFAULT_MAX_RECURSE_DEPTH,
+)
 from litellm.integrations.custom_guardrail import CustomGuardrail
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.types.utils import (
@@ -61,6 +66,7 @@ from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern
 
 MAX_KEYWORD_VALUE_GAP_WORDS: Final = 1
 GAP_WORD_TOKENIZER: Final = re.compile(r"\b\w+\b")
+SENTENCE_TERMINATORS: Final = re.compile(r"[.!?]+")
 
 
 WORD_NUMBER_MAP: Final = {
@@ -112,6 +118,22 @@ class _CategoryConfigView(TypedDict):
     category_file: str | None
 
 
+@dataclass(frozen=True, slots=True)
+class _StreamedChoiceState:
+    buffered_text: str = ""
+    yielded_masked_text_len: int = 0
+    committed_detections: tuple[ContentFilterDetection, ...] = ()
+    latest_detections: tuple[ContentFilterDetection, ...] = ()
+    next_trim_len: int = 0
+
+
+@dataclass(frozen=True, slots=True)
+class _StreamedScanPlan:
+    context_chars: int
+    exception_phrases: tuple[str, ...]
+    conditional_words: tuple[str, ...]
+
+
 class CategoryFileData(TypedDict, total=False):
     category_name: str
     description: str
@@ -976,7 +998,7 @@ class ContentFilterGuardrail(CustomGuardrail):
 
         # Split text into sentences for more precise matching
         # Simple sentence splitting on common terminators
-        sentences: Final = re.split(r"[.!?]+", text)
+        sentences: Final = SENTENCE_TERMINATORS.split(text)
 
         for category_name, config in self.conditional_categories.items():
             identifier_words = config["identifier_words"]
@@ -1950,6 +1972,81 @@ class ContentFilterGuardrail(CustomGuardrail):
                 exception_str=exception_str,
             )
 
+    def _streamed_scan_plan(self) -> _StreamedScanPlan:
+        """
+        Per-stream inputs for buffer trimming: the retained tail length (the default
+        context, widened to the longest configured keyword), the category exception
+        phrases, which suppress matches anywhere in the scanned text, and the conditional
+        category words, which only match when paired inside one sentence.
+        """
+        longest_keyword: Final = max(
+            map(len, (*self.blocked_words, *self.category_keywords, *self.always_block_category_keywords)),
+            default=0,
+        )
+        return _StreamedScanPlan(
+            context_chars=max(CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS, longest_keyword),
+            exception_phrases=tuple(
+                phrase for category in self.loaded_categories.values() for phrase in category.exceptions
+            ),
+            conditional_words=tuple(
+                word
+                for config in self.conditional_categories.values()
+                for word in (*config["identifier_words"], *config["block_words"])
+            ),
+        )
+
+    @staticmethod
+    def _cut_breaks_wider_context(buffered_text: str, head: str, tail: str, plan: _StreamedScanPlan) -> bool:
+        buffered_lower: Final = buffered_text.lower()
+        tail_lower: Final = tail.lower()
+        if any(phrase in buffered_lower and phrase not in tail_lower for phrase in plan.exception_phrases):
+            return True
+        cut_sentence: Final = (
+            SENTENCE_TERMINATORS.split(head.lower())[-1] + SENTENCE_TERMINATORS.split(tail_lower, maxsplit=1)[0]
+        )
+        return any(word in cut_sentence for word in plan.conditional_words)
+
+    def _trim_streamed_choice_buffer(
+        self, state: _StreamedChoiceState, masked_text: str, plan: _StreamedScanPlan
+    ) -> _StreamedChoiceState:
+        """
+        Bound the per-choice buffer rescanned on every streamed chunk.
+
+        Once the buffer exceeds twice the scan context, drop everything but the last
+        context-sized tail, provided no exception phrase or unfinished conditional sentence
+        would leave the buffer, the two halves mask to the same output as the whole (so no
+        match or phrase straddles the cut), and the dropped prefix has already been yielded.
+        Otherwise keep the buffer and retry once it has grown by another context length.
+
+        Detections found in the dropped prefix move to the state's committed detections.
+        """
+        if len(state.buffered_text) <= max(2 * plan.context_chars, state.next_trim_len):
+            return state
+        deferred: Final = replace(state, next_trim_len=len(state.buffered_text) + plan.context_chars)
+        head: Final = state.buffered_text[: -plan.context_chars]
+        tail: Final = state.buffered_text[-plan.context_chars :]
+        if self._cut_breaks_wider_context(state.buffered_text, head, tail, plan):
+            return deferred
+        head_detections: Final[list[ContentFilterDetection]] = []  # mutable-ok: filled by _filter_single_text
+        try:
+            masked_head: Final = self._filter_single_text(head, detections=head_detections)
+            masked_tail: Final = self._filter_single_text(tail)
+        except Exception:
+            return deferred
+        if masked_head + masked_tail != masked_text or len(masked_head) > state.yielded_masked_text_len:
+            return deferred
+        return replace(
+            state,
+            buffered_text=tail,
+            yielded_masked_text_len=state.yielded_masked_text_len - len(masked_head),
+            committed_detections=state.committed_detections + tuple(head_detections),
+            next_trim_len=0,
+        )
+
+    @staticmethod
+    def _merge_detections(detections: Sequence[ContentFilterDetection]) -> tuple[ContentFilterDetection, ...]:
+        return tuple(detection for index, detection in enumerate(detections) if detection not in detections[:index])
+
     async def async_post_call_streaming_iterator_hook(
         self,
         user_api_key_dict: UserAPIKeyAuth,
@@ -1968,10 +2065,8 @@ class ContentFilterGuardrail(CustomGuardrail):
         and the UI Request Lifecycle panel. Mirrors apply_guardrail's finally-block
         contract.
         """
-        accumulated_text_by_choice: Final[dict[int, str]] = {}
-        yielded_masked_text_len_by_choice: Final[dict[int, int]] = {}
-        latest_detections_by_choice: Final[dict[int, list[ContentFilterDetection]]] = {}
-        buffer_size: Final = 50  # Increased buffer to catch patterns split across many chunks
+        state_by_choice: Final[dict[int, _StreamedChoiceState]] = {}
+        plan: Final = self._streamed_scan_plan()
 
         start_time: Final = datetime.now()
         scan_seconds: float = 0.0  # rebind-ok: accumulates per-chunk scan time across the stream
@@ -1997,69 +2092,60 @@ class ContentFilterGuardrail(CustomGuardrail):
 
                         content = getattr(choice.delta, "content", None)
                         is_final = bool(getattr(choice, "finish_reason", None))
-                        if isinstance(content, str) and content:
-                            accumulated_text_by_choice[choice_index] = (
-                                accumulated_text_by_choice.get(choice_index, "") + content
-                            )
-                        elif not is_final:
+                        new_content = content if isinstance(content, str) else ""
+                        if not new_content and not is_final:
                             continue
 
-                        text_to_check = accumulated_text_by_choice.get(choice_index, "")
-                        if not text_to_check:
+                        previous_state = state_by_choice.get(choice_index, _StreamedChoiceState())
+                        buffered_text = previous_state.buffered_text + new_content
+                        if not buffered_text:
                             continue
 
                         # Add a space at the end if it's the final chunk to trigger word boundaries (\b)
-                        text_to_scan = text_to_check + (" " if is_final else "")
+                        text_to_scan = buffered_text + (" " if is_final else "")
                         choice_detections: list[ContentFilterDetection] = []
                         scan_started = time.perf_counter()
 
                         try:
-                            # _filter_single_text scans the whole accumulated
-                            # choice buffer every chunk, so previous-chunk
-                            # matches are guaranteed to be re-found. Keeping
-                            # only each choice's latest scan avoids duplicate
-                            # detections in the final log row.
                             masked_text = self._filter_single_text(text_to_scan, detections=choice_detections)
                             if is_final and masked_text.endswith(" "):
                                 masked_text = masked_text[:-1]
-                            latest_detections_by_choice[choice_index] = choice_detections
+                            latest_detections = tuple(choice_detections)
                         except HTTPException:
-                            latest_detections_by_choice[choice_index] = choice_detections
+                            state_by_choice[choice_index] = replace(
+                                previous_state, latest_detections=tuple(choice_detections)
+                            )
                             raise
                         except Exception as e:
                             verbose_proxy_logger.error("ContentFilterGuardrail: Error in masking: %s", e)
                             masked_text = text_to_scan  # Fallback to current text
+                            latest_detections = previous_state.latest_detections
                         finally:
                             scan_seconds += time.perf_counter() - scan_started
 
-                        # Determine how much can be safely yielded
+                        safe_to_yield_len = max(
+                            previous_state.yielded_masked_text_len,
+                            len(masked_text) - (0 if is_final else CONTENT_FILTER_STREAMING_HOLDBACK_CHARS),
+                        )
+                        choice.delta.content = masked_text[previous_state.yielded_masked_text_len : safe_to_yield_len]
+                        next_state = replace(
+                            previous_state,
+                            buffered_text=buffered_text,
+                            yielded_masked_text_len=safe_to_yield_len,
+                            latest_detections=latest_detections,
+                        )
                         if is_final:
-                            safe_to_yield_len = len(masked_text)
-                        else:
-                            safe_to_yield_len = max(0, len(masked_text) - buffer_size)
+                            state_by_choice[choice_index] = next_state
+                            continue
 
-                        yielded_masked_text_len = yielded_masked_text_len_by_choice.get(choice_index, 0)
-                        if safe_to_yield_len > yielded_masked_text_len:
-                            new_masked_content = masked_text[yielded_masked_text_len:safe_to_yield_len]
-                            choice.delta.content = new_masked_content
-                            yielded_masked_text_len_by_choice[choice_index] = safe_to_yield_len
-                        else:
-                            # Hold content by yielding empty content on this choice
-                            # while preserving chunk metadata and other choices.
-                            choice.delta.content = ""
+                        trim_started = time.perf_counter()
+                        state_by_choice[choice_index] = self._trim_streamed_choice_buffer(next_state, masked_text, plan)
+                        scan_seconds += time.perf_counter() - trim_started
 
                     yield item
                 else:
                     # Not a ModelResponseStream or no choices - yield as is
                     yield item
-
-            # Any remaining content (should have been handled by is_final, but just in case)
-            if any(
-                yielded_masked_text_len_by_choice.get(choice_index, 0) < len(accumulated_text)
-                for choice_index, accumulated_text in accumulated_text_by_choice.items()
-            ):
-                # We already reached the end of the generator
-                pass
         except HTTPException:
             status = "guardrail_intervened"
             raise
@@ -2070,8 +2156,8 @@ class ContentFilterGuardrail(CustomGuardrail):
         finally:
             detections = [
                 detection
-                for choice_detections in latest_detections_by_choice.values()
-                for detection in choice_detections
+                for state in state_by_choice.values()
+                for detection in self._merge_detections((*state.committed_detections, *state.latest_detections))
             ]
             self._count_masked_entities(detections, masked_entity_count)
             self._log_guardrail_information(

tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py

@@ -4,6 +4,7 @@ Tests for the Content Filter Guardrail
 
 import json
 import os
+from typing import Final
 from unittest.mock import MagicMock
 
 import pytest
@@ -11,6 +12,10 @@ import pytest
 
 from fastapi import HTTPException
 
+from litellm.constants import (
+    CONTENT_FILTER_STREAMING_HOLDBACK_CHARS,
+    CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS,
+)
 from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import (
     ContentFilterGuardrail,
 )
@@ -22,7 +27,9 @@ from litellm.types.guardrails import (
 )
 from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import (
     ContentFilterCategoryConfig,
+    ContentFilterDetection,
 )
+from litellm.types.utils import StandardLoggingGuardrailInformation
 
 
 class TestContentFilterGuardrail:
@@ -900,6 +907,341 @@ class TestContentFilterGuardrail:
         # masked_entity_count for email is the real count, not N×.
         assert entry["masked_entity_count"].get("email") == 1
 
+    @staticmethod
+    async def _collect_streamed_text(
+        guardrail: ContentFilterGuardrail,
+        chunks: list[str],
+        metadata: dict[str, list[StandardLoggingGuardrailInformation]],
+    ) -> str:
+        from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
+
+        async def mock_stream():
+            for i, content in enumerate(chunks):
+                yield ModelResponseStream(
+                    id=f"c{i}",
+                    choices=[StreamingChoices(delta=Delta(content=content), index=0)],
+                    model="gpt-4",
+                )
+            yield ModelResponseStream(
+                id="final",
+                choices=[
+                    StreamingChoices(
+                        delta=Delta(content=""), index=0, finish_reason="stop"
+                    )
+                ],
+                model="gpt-4",
+            )
+
+        yielded: Final[list[str]] = []
+        async for chunk in guardrail.async_post_call_streaming_iterator_hook(
+            user_api_key_dict=MagicMock(),
+            response=mock_stream(),
+            request_data={"messages": [], "model": "gpt-4o", "metadata": metadata},
+        ):
+            yielded.append(chunk.choices[0].delta.content or "")
+        return "".join(yielded)
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_scans_bounded_window_per_chunk(self):
+        """
+        Regression: the streaming hook used to re-scan the whole accumulated
+        buffer on every chunk, so scan work grew quadratically with the length
+        of the response. Each scan must now cover only the new chunk plus a
+        bounded tail of what came before, without dropping any output.
+        """
+        scanned_lengths: Final[list[int]] = []
+
+        class RecordingGuardrail(ContentFilterGuardrail):
+            def _filter_single_text(
+                self,
+                text: str,
+                detections: list[ContentFilterDetection] | None = None,
+            ) -> str:
+                scanned_lengths.append(len(text))
+                return super()._filter_single_text(text, detections=detections)
+
+        guardrail: Final = RecordingGuardrail(
+            guardrail_name="test-streaming-bounded-scan",
+            patterns=[
+                ContentFilterPattern(
+                    pattern_type="prebuilt",
+                    pattern_name="email",
+                    action=ContentFilterAction.MASK,
+                )
+            ],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        chunk: Final = "Item: a plain household object description. "
+        chunks: Final = [chunk] * 200
+
+        streamed: Final = await self._collect_streamed_text(guardrail, chunks, {})
+
+        assert streamed == chunk * 200
+        assert len(chunk) * 200 > 4 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS
+        window_bound: Final = 2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + len(chunk) + 1
+        assert max(scanned_lengths) <= window_bound, (
+            f"scan input grew to {max(scanned_lengths)} chars for a "
+            f"{len(chunk)}-char chunk; expected at most {window_bound}"
+        )
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_retries_refused_cut_once_per_context_length(self):
+        """
+        A single URL that keeps growing crosses every proposed cut, so no cut is
+        ever safe. The trim check must then back off instead of adding two extra
+        scans on every chunk, and the whole URL must still come out masked.
+        """
+        scanned_lengths: Final[list[int]] = []
+
+        class RecordingGuardrail(ContentFilterGuardrail):
+            def _filter_single_text(
+                self,
+                text: str,
+                detections: list[ContentFilterDetection] | None = None,
+            ) -> str:
+                scanned_lengths.append(len(text))
+                return super()._filter_single_text(text, detections=detections)
+
+        guardrail: Final = RecordingGuardrail(
+            guardrail_name="test-streaming-refused-cut-backoff",
+            patterns=[
+                ContentFilterPattern(
+                    pattern_type="prebuilt",
+                    pattern_name="url",
+                    action=ContentFilterAction.MASK,
+                )
+            ],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        text: Final = "See https://example.com/" + "a" * (8 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS) + " now."
+        chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)]
+
+        streamed: Final = await self._collect_streamed_text(guardrail, chunks, {})
+        streamed_scans: Final = len(scanned_lengths)
+
+        full_scan: Final = await guardrail.apply_guardrail(
+            inputs={"texts": [text]}, request_data={}, input_type="response"
+        )
+        assert streamed == full_scan["texts"][0] == "See [URL_REDACTED] now."
+        extra_scans: Final = streamed_scans - len(chunks)
+        assert extra_scans <= 2 * (len(text) // CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS), (
+            f"{extra_scans} scans beyond one per chunk for {len(chunks)} chunks; the refused cut must back off"
+        )
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_blocks_match_longer_than_holdback_across_chunks(
+        self,
+    ):
+        """
+        A blocked phrase longer than the holdback window arrives in small chunks,
+        so its start has already been yielded before its end shows up. The scan
+        still has to see the whole phrase and block.
+        """
+        phrase: Final = "alpha bravo charlie delta echo foxtrot golf hotel india juliet kilo lima"
+        assert len(phrase) > CONTENT_FILTER_STREAMING_HOLDBACK_CHARS
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="test-streaming-long-block",
+            blocked_words=[BlockedWord(keyword=phrase, action=ContentFilterAction.BLOCK)],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        text: Final = "Here is the codeword list: " + phrase + " and that is all."
+        chunks: Final = [text[i : i + 4] for i in range(0, len(text), 4)]
+        metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {}
+
+        with pytest.raises(HTTPException) as exc_info:
+            await self._collect_streamed_text(guardrail, chunks, metadata)
+
+        assert exc_info.value.detail["keyword"] == phrase
+        entry: Final = metadata["standard_logging_guardrail_information"][0]
+        assert entry["guardrail_status"] == "guardrail_intervened"
+        assert [d["keyword"] for d in entry["guardrail_response"]] == [phrase]
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_blocks_keyword_longer_than_scan_context(self):
+        """
+        A blocked keyword longer than the default retained context arrives after
+        enough text that the buffer has already been trimmed at least once. The
+        retained tail must be wide enough that the keyword's start is still in the
+        buffer when its end arrives, so the stream is blocked.
+        """
+        phrase: Final = " ".join(f"token{i:03d}" for i in range(80))
+        assert len(phrase) > CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="test-streaming-keyword-wider-than-context",
+            blocked_words=[BlockedWord(keyword=phrase, action=ContentFilterAction.BLOCK)],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        filler: Final = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23)
+        text: Final = filler + phrase + " and that is all."
+        chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)]
+        metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {}
+
+        with pytest.raises(HTTPException) as exc_info:
+            await self._collect_streamed_text(guardrail, chunks, metadata)
+
+        assert exc_info.value.detail["keyword"] == phrase
+        entry: Final = metadata["standard_logging_guardrail_information"][0]
+        assert entry["guardrail_status"] == "guardrail_intervened"
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_keeps_early_exception_phrase_suppressing_later_keyword(self):
+        """
+        Category exception phrases suppress category matches anywhere in the
+        scanned text. An exception phrase at the start of a long response must keep
+        suppressing a category keyword that arrives long after the buffer would
+        otherwise have been trimmed, exactly as one scan of the full text does.
+        """
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="test-streaming-exception-context",
+            categories=[{"category": "harmful_self_harm", "enabled": True, "action": "BLOCK"}],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        exception_phrase: Final = guardrail.loaded_categories["harmful_self_harm"].exceptions[0]
+        keyword: Final = next(iter(guardrail.category_keywords))
+        filler: Final = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23)
+        text: Final = f"Resources on {exception_phrase} matter. {filler}Someone said {keyword} in a novel."
+        chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)]
+
+        streamed: Final = await self._collect_streamed_text(guardrail, chunks, {})
+
+        full_scan: Final = await guardrail.apply_guardrail(
+            inputs={"texts": [text]}, request_data={}, input_type="response"
+        )
+        assert streamed == full_scan["texts"][0] == text
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_blocks_conditional_pair_split_by_long_sentence(self):
+        """
+        Conditional categories block an identifier word and a block word that
+        share one sentence. When the sentence runs longer than the retained
+        context, the identifier at its start must still be in the buffer when the
+        block word arrives, so the stream is blocked like a scan of the full text.
+        """
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="test-streaming-conditional-context",
+            categories=[{"category": "harmful_child_safety", "enabled": True, "action": "BLOCK"}],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        conditional: Final = guardrail.conditional_categories["harmful_child_safety"]
+        identifier, block_word = conditional["identifier_words"][0], conditional["block_words"][-1]
+        filler: Final = "and then more plain words " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 26)
+        text: Final = f"In this chapter the {identifier} {filler}shared an {block_word} moment. The end."
+        chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)]
+        metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {}
+
+        with pytest.raises(HTTPException):
+            await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response")
+        with pytest.raises(HTTPException) as exc_info:
+            await self._collect_streamed_text(guardrail, chunks, metadata)
+
+        assert "harmful_child_safety" in str(exc_info.value.detail)
+        entry: Final = metadata["standard_logging_guardrail_information"][0]
+        assert entry["guardrail_status"] == "guardrail_intervened"
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_blocks_conditional_identifier_straddling_cut(self):
+        """
+        The buffer is cut at a character offset, so a conditional identifier word
+        can sit half in the dropped head and half in the retained tail. That cut
+        must be refused: otherwise the block word arriving later in the same
+        sentence finds no identifier and the stream passes where a scan of the
+        full text blocks.
+        """
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="test-streaming-conditional-straddle",
+            categories=[{"category": "harmful_child_safety", "enabled": True, "action": "BLOCK"}],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        conditional: Final = guardrail.conditional_categories["harmful_child_safety"]
+        identifier, block_word = conditional["identifier_words"][0], conditional["block_words"][-1]
+        chunk_size: Final = 16
+        first_cut: Final = (
+            2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // chunk_size + 1
+        ) * chunk_size - CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS
+        prefix: Final = ("plain words " * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS)[: first_cut - 2]
+        filler: Final = "and then more plain words " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 26)
+        text: Final = f"{prefix}{identifier} {filler}shared an {block_word} moment. The end."
+        assert text[first_cut - 2 : first_cut - 2 + len(identifier)] == identifier
+        chunks: Final = [text[i : i + chunk_size] for i in range(0, len(text), chunk_size)]
+        metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {}
+
+        with pytest.raises(HTTPException):
+            await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response")
+        with pytest.raises(HTTPException) as exc_info:
+            await self._collect_streamed_text(guardrail, chunks, metadata)
+
+        assert "harmful_child_safety" in str(exc_info.value.detail)
+        entry: Final = metadata["standard_logging_guardrail_information"][0]
+        assert entry["guardrail_status"] == "guardrail_intervened"
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_masks_every_email_in_long_stream_and_logs_once(
+        self,
+    ):
+        """
+        A response made of nothing but emails, several times longer than the
+        rescanned buffer, must come out as nothing but redaction tags, and the log
+        must carry one email detection, matching what a single scan of the full
+        text reports. Wherever the buffer is cut, an email sits on the cut, so
+        dropping text without checking that the cut leaves the masked output
+        unchanged corrupts the stream.
+        """
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="test-streaming-many-emails",
+            patterns=[
+                ContentFilterPattern(
+                    pattern_type="prebuilt",
+                    pattern_name="email",
+                    action=ContentFilterAction.MASK,
+                )
+            ],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        emails: Final = [f"user{i:03d}@example.com" for i in range(200)]
+        text: Final = " ".join(emails)
+        assert len(text) > 4 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS
+        chunks: Final = [text[i : i + 3] for i in range(0, len(text), 3)]
+        metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {}
+
+        streamed: Final = await self._collect_streamed_text(guardrail, chunks, metadata)
+
+        assert streamed == " ".join(["[EMAIL_REDACTED]"] * len(emails))
+        entry: Final = metadata["standard_logging_guardrail_information"][0]
+        assert entry["guardrail_status"] == "success"
+        assert [d["pattern_name"] for d in entry["guardrail_response"]] == ["email"]
+        assert entry["masked_entity_count"] == {"email": 1}
+
+    @pytest.mark.asyncio
+    async def test_streaming_hook_logs_detection_masked_long_before_stream_end(self):
+        """
+        An email at the start of a long response is masked and then falls out of
+        the rescanned buffer well before the stream ends. The final log entry must
+        still report it, as a scan of the full text would.
+        """
+        guardrail: Final = ContentFilterGuardrail(
+            guardrail_name="test-streaming-early-detection",
+            patterns=[
+                ContentFilterPattern(
+                    pattern_type="prebuilt",
+                    pattern_name="email",
+                    action=ContentFilterAction.MASK,
+                )
+            ],
+            event_hook=GuardrailEventHooks.post_call,
+        )
+        filler: Final = "filler text " * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS
+        text: Final = f"Contact one@example.com for details. {filler}"
+        chunks: Final = [text[i : i + 40] for i in range(0, len(text), 40)]
+        metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {}
+
+        streamed: Final = await self._collect_streamed_text(guardrail, chunks, metadata)
+
+        assert streamed == text.replace("one@example.com", "[EMAIL_REDACTED]")
+        entry: Final = metadata["standard_logging_guardrail_information"][0]
+        assert entry["guardrail_status"] == "success"
+        assert [d["pattern_name"] for d in entry["guardrail_response"]] == ["email"]
+        assert entry["masked_entity_count"] == {"email": 1}
+
     def test_init_with_plain_dicts(self):
         """
         Test initialization with plain dicts (DB format).