diff --git a/litellm/constants.py b/litellm/constants.py index 745a4d9294e..ce5b65080ee 100644 --- a/litellm/constants.py +++ b/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 diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index 1e684c514de..092e8eaafa1 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/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( diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index 130b0da000b..d0ad068aeb4 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/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).