From b3d9ba9e7bc745b8a4c01f8d7c2add00959f2f38 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 18 Sep 2026 22:59:52 -0700 Subject: [PATCH 1/3] fix(policy_engine): deliver guardrail text rewrites on multi-choice, unfinished, and envelope-less streams Post-call pipeline rewrites on buffered streams failed open on three shapes: chat streams with n > 1 (the rebuilt response collapsed every choice into index 0), streams that ended without a finish marker, and Responses streams whose final event carried no response envelope. The chat handler now rebuilds the ended stream one choice index at a time and writes each choice's rewrite back to that choice's buffered deltas. The Anthropic handler writes an unended stream's rewrite across its text deltas. The Responses handler spreads an envelope-less rewrite over the buffered output_text events, still failing open when a scanned event cannot be placed. Tool-call rewrites on n > 1 chat streams keep failing open. --- .../chat/guardrail_translation/handler.py | 11 +- .../chat/guardrail_translation/handler.py | 96 +++++++++++------ .../guardrail_translation/handler.py | 64 +++++++++-- .../test_anthropic_guardrail_handler.py | 23 ++-- .../test_openai_guardrail_handler.py | 73 +++++++++++-- ...test_openai_responses_guardrail_handler.py | 100 ++++++++++++++---- .../policy_engine/test_pipeline_executor.py | 36 +++++++ 7 files changed, 311 insertions(+), 92 deletions(-) diff --git a/litellm/llms/anthropic/chat/guardrail_translation/handler.py b/litellm/llms/anthropic/chat/guardrail_translation/handler.py index 373435fa4ee..a2fbf612204 100644 --- a/litellm/llms/anthropic/chat/guardrail_translation/handler.py +++ b/litellm/llms/anthropic/chat/guardrail_translation/handler.py @@ -1253,10 +1253,9 @@ class AnthropicMessagesHandler(BaseTranslation): Process output streaming response by applying guardrails to text content. Get the string so far, check the apply guardrail to the string so far, and return the list of responses so far. - With ``deliver_ended_stream_rewrites``, an ended stream whose guardrail rewrote the text gets the rewrite - written back across the buffered chunks (full rewritten text in the first ``text_delta``, the rest blanked); - a rewrite on a stream that never reported a ``stop_reason`` has no write-back and is reported as - undeliverable, so the pipeline executor discards it and releases the original chunks. + With ``deliver_ended_stream_rewrites``, a stream whose guardrail rewrote the text gets the rewrite + written back across the buffered chunks (full rewritten text in the first ``text_delta``, the rest blanked), + whether or not the stream ever reported a ``stop_reason``. """ from litellm.integrations.custom_guardrail import ModifyResponseException @@ -1354,9 +1353,7 @@ class AnthropicMessagesHandler(BaseTranslation): raise unended_texts: Final = _guardrailed_inputs.get("texts") if deliver_ended_stream_rewrites and unended_texts and tuple(unended_texts) != (string_so_far,): - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - - raise UndeliverableStreamRewrite(guardrail_to_apply.guardrail_name or "unknown") + self._write_ended_stream_text_rewrite(responses_so_far, unended_texts[0]) return responses_so_far def _prepare_request_data( diff --git a/litellm/llms/openai/chat/guardrail_translation/handler.py b/litellm/llms/openai/chat/guardrail_translation/handler.py index f85d238484e..b245394e1c0 100644 --- a/litellm/llms/openai/chat/guardrail_translation/handler.py +++ b/litellm/llms/openai/chat/guardrail_translation/handler.py @@ -18,6 +18,7 @@ import json import time import uuid from collections.abc import Sequence +from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Union, cast from typing_extensions import NotRequired, ReadOnly, TypedDict @@ -651,10 +652,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation): """Ended-stream path: rebuild the full response, run the non-streaming output guardrail against it, and (when opted in) write any text or tool-call rewrite back across the buffered chunks.""" - model_response: Final = cast( - ModelResponse, - stream_chunk_builder(chunks=responses_so_far, logging_obj=litellm_logging_obj), - ) + model_response: Final = self._rebuild_ended_stream_per_choice(responses_so_far, litellm_logging_obj) pre_guardrail_texts: Final = self._string_choice_contents(model_response) pre_guardrail_tool_calls: Final = self._function_tool_call_shapes(model_response) await self.process_output_response( @@ -666,18 +664,61 @@ class OpenAIChatCompletionsHandler(BaseTranslation): ) if not deliver_ended_stream_rewrites: return - guardrail_name: Final = guardrail_to_apply.guardrail_name or "unknown" await self._write_ended_stream_text_rewrites( responses_so_far=responses_so_far, guardrailed_response=model_response, pre_guardrail_texts=pre_guardrail_texts, - guardrail_name=guardrail_name, ) self._write_ended_stream_tool_call_rewrites( responses_so_far=responses_so_far, guardrailed_response=model_response, pre_guardrail_tool_calls=pre_guardrail_tool_calls, - guardrail_name=guardrail_name, + guardrail_name=guardrail_to_apply.guardrail_name or "unknown", + ) + + @staticmethod + def _rebuild_ended_stream_per_choice( + responses_so_far: Sequence["ModelResponseStream"], + litellm_logging_obj: "LiteLLMLoggingObj | None", + ) -> "ModelResponse": + """``stream_chunk_builder`` folds every choice of a stream into one index-0 + choice, so the stream is rebuilt one choice index at a time (every chunk + kept, its choices narrowed to that index, so usage-only chunks still + count) and the rebuilt choices are stitched into one response, each + carrying the index the stream gave it.""" + choice_indices: Final = tuple( + sorted(frozenset(choice.index for response in responses_so_far for choice in response.choices)) + ) + rebuilt_by_index: Final = tuple( + ( + index, + cast( + ModelResponse, + stream_chunk_builder( + chunks=[ # mutable-ok: callee takes a list + response.model_copy( + update=MappingProxyType( + {"choices": tuple(choice for choice in response.choices if choice.index == index)} + ) + ) + for response in responses_so_far + ], + logging_obj=litellm_logging_obj, + ), + ), + ) + for index in choice_indices + ) + (_, base_response), *_ = rebuilt_by_index + return base_response.model_copy( + update=MappingProxyType( + { + "choices": tuple( + rebuilt.choices[0].model_copy(update=MappingProxyType({"index": index})) + for index, rebuilt in rebuilt_by_index + ) + } + ) ) def build_stream_error_items( @@ -1058,39 +1099,28 @@ class OpenAIChatCompletionsHandler(BaseTranslation): responses_so_far: list["ModelResponseStream"], # mutable-ok: rewrites the caller's buffered chunks in place guardrailed_response: "ModelResponse", pre_guardrail_texts: tuple[str | None, ...], - guardrail_name: str, ) -> None: """Write ended-stream guardrail text rewrites back across the buffered - chunks: the full rewritten text lands in the choice's first - content-carrying chunk and the rest are blanked, the same shape the - in-flight write-back uses. Chunks carrying only finish_reason or usage - stay untouched. A rewrite on a stream carrying more than one distinct - choice index is reported as undeliverable, so the pipeline executor - discards it and releases the original chunks.""" + chunks, one rewrite per rebuilt choice index: the full rewritten text + lands in that choice's first content-carrying chunk and the rest are + blanked, the same shape the in-flight write-back uses. Chunks carrying + only finish_reason or usage stay untouched.""" post_guardrail_texts: Final = self._string_choice_contents(guardrailed_response) - changed: Final = tuple( - after - for before, after in zip(pre_guardrail_texts, post_guardrail_texts) - if before is not None and after is not None and after != before + rewrites_by_choice: Final = MappingProxyType( + { + choice.index: after + for choice, before, after in zip( + guardrailed_response.choices, pre_guardrail_texts, post_guardrail_texts + ) + if before is not None and after is not None and after != before + } ) - if not changed: + if not rewrites_by_choice: return - stream_choice_indices: Final = frozenset( - choice.index for response in responses_so_far for choice in response.choices - ) - if len(stream_choice_indices) != 1: - # stream_chunk_builder collapses every choice into one index-0 - # choice, so a rewrite of the rebuilt response cannot be attributed - # back to a single choice on an n>1 stream: report it undeliverable - # rather than deliver the rewrite on the wrong choice - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - - raise UndeliverableStreamRewrite(guardrail_name) - target_choice_index: Final = next(iter(stream_choice_indices)) await self._apply_guardrail_responses_to_output_streaming( responses=responses_so_far, - guardrailed_texts=list(changed), # mutable-ok: callee takes lists - task_mappings=[(target_choice_index, None) for _ in changed], # mutable-ok: callee takes lists + guardrailed_texts=list(rewrites_by_choice.values()), # mutable-ok: callee takes lists + task_mappings=[(index, None) for index in rewrites_by_choice], # mutable-ok: callee takes lists ) @staticmethod diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py index 982bb137a30..e3e53f9b3dc 100644 --- a/litellm/llms/openai/responses/guardrail_translation/handler.py +++ b/litellm/llms/openai/responses/guardrail_translation/handler.py @@ -209,6 +209,7 @@ _TOOL_CALL_PAYLOAD_EVENT_TYPES: Final = _TOOL_CALL_PAYLOAD_DELTA_EVENT_TYPES | f _TOOL_CALL_PAYLOAD_DONE_EVENT_FIELDS ) _OUTPUT_ITEM_EVENT_TYPES: Final = frozenset({"response.output_item.added", "response.output_item.done"}) +_OUTPUT_TEXT_EVENT_TYPES: Final = frozenset({"response.output_text.delta", "response.output_text.done"}) _PATCHABLE_ITEM_FIELDS: Final[Mapping[str, str]] = MappingProxyType( {"function_call_output": "output", "message": "content"} ) @@ -832,9 +833,10 @@ class OpenAIResponsesHandler(BaseTranslation): (``response.output_text.delta`` / ``.done``, ``response.content_part.done``, ``response.output_item.done``) are synced to the rewritten envelope too, so a client reading deltas sees the - rewrite instead of the raw model output; a rewrite observed where no - write-back is possible is reported as undeliverable, so the pipeline - executor discards it and releases the original events. + rewrite instead of the raw model output; a stream with no envelope + gets its rewrite spread over the buffered text events, and a rewrite + observed where no write-back is possible is reported as undeliverable, + so the pipeline executor discards it and releases the original events. """ if not responses_so_far: return responses_so_far @@ -958,10 +960,9 @@ class OpenAIResponsesHandler(BaseTranslation): return responses_so_far # ------------------------------------------------------------------ # - # Fallback: apply guardrail to the accumulated text string. # - # No structured write-back is possible here; guardrails that only # - # need to block/flag (not rewrite) still work correctly, and a # - # rewrite a caller expects delivered is reported undeliverable. # + # Fallback: apply guardrail to the accumulated text string. With no # + # envelope to rewrite, a rewrite a caller expects delivered is spread # + # over the buffered text events instead. # # ------------------------------------------------------------------ # string_so_far: Final = self.get_streaming_string_so_far(responses_so_far) if string_so_far: @@ -979,11 +980,54 @@ class OpenAIResponsesHandler(BaseTranslation): ) fallback_texts: Final = fallback_outputs.get("texts") if deliver_ended_stream_rewrites and fallback_texts and tuple(fallback_texts) != (string_so_far,): - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - - raise UndeliverableStreamRewrite(guardrail_to_apply.guardrail_name or "unknown") + self._spread_text_rewrite_over_stream_events( + stream_events=responses_so_far, + rewritten_text=fallback_texts[0], + guardrail_name=guardrail_to_apply.guardrail_name or "unknown", + ) return responses_so_far + def _spread_text_rewrite_over_stream_events( + self, + stream_events: Sequence[Any], + rewritten_text: str, + guardrail_name: str, + ) -> None: + """Deliver a text rewrite on a stream with no completed envelope by + spreading it over the text parts the guardrail scanned, in stream + order: the whole rewrite on the first part and every later part + blanked, through the same sync the envelope path uses. A scanned + event the sync cannot place (one that is not an ``output_text`` delta + or done, or lacks integer ``output_index`` / ``content_index``) makes + the rewrite undeliverable, so the pipeline executor discards it and + releases the original events.""" + scanned_events: Final = tuple( + event + for event in stream_events + if isinstance(stream_item_field(event, "text"), str) or isinstance(stream_item_field(event, "delta"), str) + ) + scanned_positions: Final = tuple( + dict.fromkeys( + (stream_item_field(event, "output_index"), stream_item_field(event, "content_index")) + for event in scanned_events + ) + ) + placeable_positions: Final = tuple( + (output_index, content_index) + for output_index, content_index in scanned_positions + if isinstance(output_index, int) and isinstance(content_index, int) + ) + if len(placeable_positions) != len(scanned_positions) or any( + stream_item_field(event, "type") not in _OUTPUT_TEXT_EVENT_TYPES for event in scanned_events + ): + from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite + + raise UndeliverableStreamRewrite(guardrail_name) + self._sync_stream_events_with_rewrites( + stream_events=stream_events, + rewrites_by_position=MappingProxyType(dict(zip(placeable_positions, chain((rewritten_text,), repeat(""))))), + ) + @staticmethod def _write_event_field(event: object, field: str, value: str) -> None: if isinstance(event, dict): diff --git a/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py b/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py index eaa2c4e8b9a..d47bd770671 100644 --- a/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py +++ b/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py @@ -445,19 +445,22 @@ class TestAnthropicMessagesHandlerStreamingOutputProcessing: assert chunks == original @pytest.mark.asyncio - async def test_unended_stream_rewrite_with_delivery_expected_fails_closed(self): - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - + async def test_unended_stream_rewrite_with_delivery_expected_lands_in_the_buffered_deltas(self): handler = AnthropicMessagesHandler() chunks = self._ended_sse_chunks()[:-2] - with pytest.raises(UndeliverableStreamRewrite): - await handler.process_output_streaming_response( - responses_so_far=chunks, - guardrail_to_apply=self._masking_guardrail(), - litellm_logging_obj=MagicMock(), - deliver_ended_stream_rewrites=True, - ) + result = await handler.process_output_streaming_response( + responses_so_far=chunks, + guardrail_to_apply=self._masking_guardrail(), + litellm_logging_obj=MagicMock(), + deliver_ended_stream_rewrites=True, + ) + + assert result is chunks + assert self._delta_texts(chunks) == ["hello [MASKED]", ""] + raw = b"".join(chunks).decode() + assert "event: message_start" in raw and "event: content_block_stop" in raw + assert "event: message_stop" not in raw @pytest.mark.asyncio async def test_unended_stream_without_rewrite_is_released_with_delivery_expected(self): diff --git a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py index e4e9f5d33db..1bc57c987fa 100644 --- a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py +++ b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py @@ -1262,20 +1262,75 @@ class TestOpenAIChatCompletionsHandlerStreamingOutput: return MaskWorld(guardrail_name="test-mask") @pytest.mark.asyncio - async def test_deliver_ended_stream_rewrite_on_multi_choice_stream_fails_closed(self): - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - + async def test_deliver_ended_stream_rewrite_lands_on_the_rewritten_choice_only(self): handler = OpenAIChatCompletionsHandler() chunks = self._two_choice_stream_chunks() - with pytest.raises(UndeliverableStreamRewrite): - await handler.process_output_streaming_response( - responses_so_far=chunks, - guardrail_to_apply=self._world_masking_guardrail(), - litellm_logging_obj=None, - deliver_ended_stream_rewrites=True, + result = await handler.process_output_streaming_response( + responses_so_far=chunks, + guardrail_to_apply=self._world_masking_guardrail(), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert result is chunks + assert [(c.choices[0].index, c.choices[0].delta.content) for c in chunks] == [ + (0, "safe "), + (1, "hello [MASKED]"), + (0, "text"), + (1, ""), + ] + assert [c.choices[0].finish_reason for c in chunks] == [None, None, "stop", "stop"] + + @pytest.mark.asyncio + async def test_deliver_ended_stream_rewrites_each_choice_with_its_own_text(self): + handler = OpenAIChatCompletionsHandler() + guardrail = MockGuardrail(guardrail_name="test") + chunks = self._two_choice_stream_chunks() + + await handler.process_output_streaming_response( + responses_so_far=chunks, + guardrail_to_apply=guardrail, + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert guardrail.last_inputs["texts"] == ["safe text", "hello world"] + assert [(c.choices[0].index, c.choices[0].delta.content) for c in chunks] == [ + (0, "SAFE TEXT"), + (1, "HELLO WORLD"), + (0, ""), + (1, ""), + ] + + @pytest.mark.asyncio + async def test_deliver_rewrite_on_unfinished_stream_lands_in_the_buffered_deltas(self): + from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices + + handler = OpenAIChatCompletionsHandler() + + def chunk(content: str) -> ModelResponseStream: + return ModelResponseStream( + id="chatcmpl-123", + created=1234567890, + model="gpt-4", + object="chat.completion.chunk", + choices=[StreamingChoices(index=0, delta=Delta(content=content), finish_reason=None)], ) + chunks = [chunk("hello "), chunk("world")] + + result = await handler.process_output_streaming_response( + responses_so_far=chunks, + guardrail_to_apply=self._world_masking_guardrail(), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert result is chunks + assert [c.choices[0].delta.content for c in chunks] == ["hello [MASKED]", ""] + assert [c.choices[0].finish_reason for c in chunks] == [None, None] + @staticmethod def _two_choice_tool_call_stream_chunks() -> list: from litellm.types.utils import ( diff --git a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py index d461b939553..872b2e1a3d5 100644 --- a/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py +++ b/tests/test_litellm/llms/openai/responses/test_openai_responses_guardrail_handler.py @@ -1747,33 +1747,82 @@ class TestOpenAIResponsesHandlerStreamingOutputProcessing: assert events[5]["response"]["output"][0]["content"][0]["text"] == "hello [MASKED]" @pytest.mark.asyncio - async def test_fallback_rewrite_with_delivery_expected_fails_closed(self): - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - + async def test_fallback_rewrite_with_delivery_expected_lands_in_the_delta_and_done_events(self): handler = OpenAIResponsesHandler() events = [ {"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "hello "}, {"type": "response.output_text.done", "output_index": 0, "content_index": 0, "text": "hello world"}, ] - with pytest.raises(UndeliverableStreamRewrite): - await handler.process_output_streaming_response( - responses_so_far=events, - guardrail_to_apply=self._masking_guardrail(), - litellm_logging_obj=None, - deliver_ended_stream_rewrites=True, - ) + result = await handler.process_output_streaming_response( + responses_so_far=events, + guardrail_to_apply=self._masking_guardrail(), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert result is events + assert events[0]["delta"] == "hello [MASKED]" + assert events[1]["text"] == "hello [MASKED]" @pytest.mark.asyncio - async def test_fallback_delta_only_rewrite_with_delivery_expected_fails_closed(self): - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - + async def test_fallback_delta_only_rewrite_with_delivery_expected_spreads_over_the_deltas(self): handler = OpenAIResponsesHandler() events = [ {"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "hello "}, {"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "world"}, ] + result = await handler.process_output_streaming_response( + responses_so_far=events, + guardrail_to_apply=self._masking_guardrail(), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert result is events + assert [event["delta"] for event in events] == ["hello [MASKED]", ""] + + @pytest.mark.asyncio + async def test_fallback_rewrite_across_parts_lands_whole_on_the_first_part(self): + handler = OpenAIResponsesHandler() + events = [ + {"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "hello "}, + {"type": "response.output_text.done", "output_index": 0, "content_index": 0, "text": "hello "}, + {"type": "response.output_text.delta", "output_index": 1, "content_index": 0, "delta": "wor"}, + {"type": "response.output_text.delta", "output_index": 1, "content_index": 0, "delta": "ld"}, + ] + guardrail = MockRecordingGuardrail(guardrail_name="test") + + await handler.process_output_streaming_response( + responses_so_far=events, + guardrail_to_apply=guardrail, + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + assert [inputs.get("texts") for inputs in guardrail.seen_inputs] == [["hello world"]] + + await handler.process_output_streaming_response( + responses_so_far=events, + guardrail_to_apply=self._masking_guardrail(), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert events[0]["delta"] == "hello [MASKED]" + assert events[1]["text"] == "hello [MASKED]" + assert [event["delta"] for event in events[2:]] == ["", ""] + + @pytest.mark.asyncio + async def test_fallback_rewrite_over_an_unplaceable_scanned_event_fails_open(self): + from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite + + handler = OpenAIResponsesHandler() + events = [ + {"type": "response.reasoning_summary_text.delta", "output_index": 0, "summary_index": 0, "delta": "hello "}, + {"type": "response.output_text.delta", "output_index": 1, "content_index": 0, "delta": "world"}, + ] + with pytest.raises(UndeliverableStreamRewrite): await handler.process_output_streaming_response( responses_so_far=events, @@ -1781,21 +1830,26 @@ class TestOpenAIResponsesHandlerStreamingOutputProcessing: litellm_logging_obj=None, deliver_ended_stream_rewrites=True, ) + assert [event["delta"] for event in events] == ["hello ", "world"] @pytest.mark.asyncio - async def test_output_item_done_last_rewrite_with_delivery_expected_fails_closed(self): - from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite - + async def test_output_item_done_last_rewrite_with_delivery_expected_syncs_every_text_event(self): handler = OpenAIResponsesHandler() events = self._ended_stream_events()[:-1] - with pytest.raises(UndeliverableStreamRewrite): - await handler.process_output_streaming_response( - responses_so_far=events, - guardrail_to_apply=self._masking_guardrail(), - litellm_logging_obj=None, - deliver_ended_stream_rewrites=True, - ) + result = await handler.process_output_streaming_response( + responses_so_far=events, + guardrail_to_apply=self._masking_guardrail(), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert result is events + assert events[0]["delta"] == "hello [MASKED]" + assert events[1]["delta"] == "" + assert events[2]["text"] == "hello [MASKED]" + assert events[3]["part"]["text"] == "hello [MASKED]" + assert events[4]["item"]["content"][0]["text"] == "hello [MASKED]" @pytest.mark.asyncio async def test_output_item_done_last_scans_text_with_delivery_expected(self): diff --git a/tests/test_litellm/proxy/policy_engine/test_pipeline_executor.py b/tests/test_litellm/proxy/policy_engine/test_pipeline_executor.py index 0a2641082dc..624bc3f077b 100644 --- a/tests/test_litellm/proxy/policy_engine/test_pipeline_executor.py +++ b/tests/test_litellm/proxy/policy_engine/test_pipeline_executor.py @@ -1318,6 +1318,42 @@ async def test_streaming_step_records_guardrail_information_once_on_block(monkey assert _recorded_guardrail_statuses(result) == ["guardrail_intervened"] +def _two_choice_chat_chunks(): + from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices + + def chunk(index, content, finish_reason=None): + return ModelResponseStream( + id="chatcmpl-123", + created=1234567890, + model="gpt-4", + object="chat.completion.chunk", + choices=[StreamingChoices(index=index, delta=Delta(content=content), finish_reason=finish_reason)], + ) + + return [chunk(0, "pers"), chunk(1, "pers"), chunk(0, "immon", "stop"), chunk(1, "immon", "stop")] + + +@pytest.mark.asyncio +async def test_streaming_step_delivers_text_rewrites_on_every_choice_of_a_chat_stream(monkeypatch, caplog): + from litellm.llms.openai.chat.guardrail_translation.handler import OpenAIChatCompletionsHandler + + monkeypatch.setattr(litellm, "callbacks", [_TextReturningGuardrail(["[MASKED]", "[MASKED]"])]) + chunks = _two_choice_chat_chunks() + + with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"): + result = await _run_streaming_step(OpenAIChatCompletionsHandler(), chunks) + + assert result.terminal_action == "allow" + assert not any("discarded" in record.getMessage() for record in caplog.records) + assert [(c.choices[0].index, c.choices[0].delta.content) for c in chunks] == [ + (0, "[MASKED]"), + (1, "[MASKED]"), + (0, ""), + (1, ""), + ] + assert result.modified_data["metadata"]["applied_guardrails"] == ["masker"] + + @pytest.mark.asyncio async def test_streaming_step_restores_chunks_when_translation_refuses_the_rewrite(monkeypatch, caplog): monkeypatch.setattr(litellm, "callbacks", [_TextReturningGuardrail(["hello [MASKED]"])]) From cb80e8773ef2c4c92040a62b618db9f0591e776d Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 18 Sep 2026 23:24:49 -0700 Subject: [PATCH 2/3] fix(policy_engine): fail open when an unended Messages stream has no text delta to carry the rewrite --- .../chat/guardrail_translation/handler.py | 40 ++++++++++++++----- .../test_anthropic_guardrail_handler.py | 22 ++++++++++ .../test_openai_guardrail_handler.py | 39 ++++++++++-------- 3 files changed, 76 insertions(+), 25 deletions(-) diff --git a/litellm/llms/anthropic/chat/guardrail_translation/handler.py b/litellm/llms/anthropic/chat/guardrail_translation/handler.py index a2fbf612204..b0e97150ded 100644 --- a/litellm/llms/anthropic/chat/guardrail_translation/handler.py +++ b/litellm/llms/anthropic/chat/guardrail_translation/handler.py @@ -1311,7 +1311,11 @@ class AnthropicMessagesHandler(BaseTranslation): and guardrailed_texts and guardrailed_texts[0] != string_so_far ): - self._write_ended_stream_text_rewrite(responses_so_far, guardrailed_texts[0]) + self._write_ended_stream_text_rewrite( + responses_so_far, + guardrailed_texts[0], + guardrail_name=guardrail_to_apply.guardrail_name or "unknown", + ) if deliver_ended_stream_rewrites: returned_tool_calls: Final = _guardrailed_inputs.get("tool_calls") self._write_ended_stream_tool_call_rewrites( @@ -1353,7 +1357,11 @@ class AnthropicMessagesHandler(BaseTranslation): raise unended_texts: Final = _guardrailed_inputs.get("texts") if deliver_ended_stream_rewrites and unended_texts and tuple(unended_texts) != (string_so_far,): - self._write_ended_stream_text_rewrite(responses_so_far, unended_texts[0]) + self._write_ended_stream_text_rewrite( + responses_so_far, + unended_texts[0], + guardrail_name=guardrail_to_apply.guardrail_name or "unknown", + ) return responses_so_far def _prepare_request_data( @@ -1447,26 +1455,40 @@ class AnthropicMessagesHandler(BaseTranslation): inputs["model"] = response_model return inputs - @staticmethod + @classmethod def _write_ended_stream_text_rewrite( + cls, responses_so_far: MutableSequence[object], # mutable-ok: rewrites the caller's buffered chunks in place rewritten_text: str, + guardrail_name: str, ) -> None: """Deliver an ended-stream guardrail text rewrite by rewriting the buffered chunks in place: the first ``text_delta`` carries the full rewritten text and every later one is blanked, leaving the surrounding - message and content-block framing untouched.""" + message and content-block framing untouched. A buffer with no + ``text_delta`` has nowhere to carry the rewrite, so the pipeline + executor discards it and releases the original chunks.""" + + def is_text_delta(event: Mapping[str, object]) -> bool: + delta: Final = event.get("delta") + return ( + event.get("type") == "content_block_delta" + and isinstance(delta, Mapping) + and delta.get("type") == "text_delta" + ) + + if not any(is_text_delta(event) for item in responses_so_far for event in cls._iter_sse_events(item)): + from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite + + raise UndeliverableStreamRewrite(guardrail_name) replacements: Final = chain((rewritten_text,), repeat("")) def rewrite_text_delta(event: Mapping[str, object]) -> _SSEFieldRewrite | None: - delta: Final = event.get("delta") - if event.get("type") != "content_block_delta" or not isinstance(delta, Mapping): - return None - if delta.get("type") != "text_delta": + if not is_text_delta(event): return None return _SSEFieldRewrite("delta", "text", next(replacements)) - AnthropicMessagesHandler._rewrite_ended_stream_events(responses_so_far, rewrite_text_delta) + cls._rewrite_ended_stream_events(responses_so_far, rewrite_text_delta) @classmethod def _write_ended_stream_tool_call_rewrites( diff --git a/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py b/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py index d47bd770671..9df6009df53 100644 --- a/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py +++ b/tests/test_litellm/llms/anthropic/chat/guardrail_translation/test_anthropic_guardrail_handler.py @@ -462,6 +462,28 @@ class TestAnthropicMessagesHandlerStreamingOutputProcessing: assert "event: message_start" in raw and "event: content_block_stop" in raw assert "event: message_stop" not in raw + @pytest.mark.asyncio + async def test_unended_stream_rewrite_with_no_text_delta_to_carry_it_fails_open(self): + from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite + + class FillEmpty(CustomGuardrail): + async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None): + return {**inputs, "texts": ["[INJECTED]" for _ in inputs.get("texts", [])]} + + handler = AnthropicMessagesHandler() + chunks = self._ended_sse_chunks()[:2] + original = [bytes(chunk) for chunk in chunks] + + with pytest.raises(UndeliverableStreamRewrite): + await handler.process_output_streaming_response( + responses_so_far=chunks, + guardrail_to_apply=FillEmpty(guardrail_name="test"), + litellm_logging_obj=MagicMock(), + deliver_ended_stream_rewrites=True, + ) + + assert chunks == original + @pytest.mark.asyncio async def test_unended_stream_without_rewrite_is_released_with_delivery_expected(self): handler = AnthropicMessagesHandler() diff --git a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py index 1bc57c987fa..9c0d7134e7c 100644 --- a/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py +++ b/tests/test_litellm/llms/openai/chat/guardrail_translation/test_openai_guardrail_handler.py @@ -1304,32 +1304,39 @@ class TestOpenAIChatCompletionsHandlerStreamingOutput: ] @pytest.mark.asyncio - async def test_deliver_rewrite_on_unfinished_stream_lands_in_the_buffered_deltas(self): - from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices + async def test_deliver_ended_stream_rewrites_every_choice_when_a_usage_only_chunk_closes_the_stream(self): + from litellm.types.utils import ModelResponseStream, Usage handler = OpenAIChatCompletionsHandler() - - def chunk(content: str) -> ModelResponseStream: - return ModelResponseStream( - id="chatcmpl-123", - created=1234567890, - model="gpt-4", - object="chat.completion.chunk", - choices=[StreamingChoices(index=0, delta=Delta(content=content), finish_reason=None)], - ) - - chunks = [chunk("hello "), chunk("world")] + guardrail = MockGuardrail(guardrail_name="test") + usage_chunk = ModelResponseStream( + id="chatcmpl-123", + created=1234567890, + model="gpt-4", + object="chat.completion.chunk", + choices=[], + usage=Usage(prompt_tokens=5, completion_tokens=7, total_tokens=12), + ) + chunks = [*self._two_choice_stream_chunks(), usage_chunk] result = await handler.process_output_streaming_response( responses_so_far=chunks, - guardrail_to_apply=self._world_masking_guardrail(), + guardrail_to_apply=guardrail, litellm_logging_obj=None, deliver_ended_stream_rewrites=True, ) assert result is chunks - assert [c.choices[0].delta.content for c in chunks] == ["hello [MASKED]", ""] - assert [c.choices[0].finish_reason for c in chunks] == [None, None] + assert guardrail.last_inputs["texts"] == ["safe text", "hello world"] + assert [(c.choices[0].index, c.choices[0].delta.content) for c in chunks[:4]] == [ + (0, "SAFE TEXT"), + (1, "HELLO WORLD"), + (0, ""), + (1, ""), + ] + assert [c.choices[0].finish_reason for c in chunks[:4]] == [None, None, "stop", "stop"] + assert chunks[4].choices == [] + assert chunks[4].usage.completion_tokens == 7 @staticmethod def _two_choice_tool_call_stream_chunks() -> list: From 4fe15494327a9f080dc742397158ce020d6f1597 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 18 Sep 2026 23:53:02 -0700 Subject: [PATCH 3/3] fix(policy_engine): keep the per-choice rebuilt response's choices a list so legacy hook rewrites survive the model_dump round-trip --- .../chat/guardrail_translation/handler.py | 26 +++++----- .../openai/test_moderations.py | 5 ++ .../test_openai_moderation_streaming.py | 3 ++ .../proxy_logging/test_guardrail_pipeline.py | 49 +++++++++++++++++++ 4 files changed, 68 insertions(+), 15 deletions(-) diff --git a/litellm/llms/openai/chat/guardrail_translation/handler.py b/litellm/llms/openai/chat/guardrail_translation/handler.py index b245394e1c0..7ea98fc5ce7 100644 --- a/litellm/llms/openai/chat/guardrail_translation/handler.py +++ b/litellm/llms/openai/chat/guardrail_translation/handler.py @@ -696,11 +696,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation): ModelResponse, stream_chunk_builder( chunks=[ # mutable-ok: callee takes a list - response.model_copy( - update=MappingProxyType( - {"choices": tuple(choice for choice in response.choices if choice.index == index)} - ) - ) + OpenAIChatCompletionsHandler._narrowed_to_choice(response, index) for response in responses_so_far ], logging_obj=litellm_logging_obj, @@ -710,16 +706,16 @@ class OpenAIChatCompletionsHandler(BaseTranslation): for index in choice_indices ) (_, base_response), *_ = rebuilt_by_index - return base_response.model_copy( - update=MappingProxyType( - { - "choices": tuple( - rebuilt.choices[0].model_copy(update=MappingProxyType({"index": index})) - for index, rebuilt in rebuilt_by_index - ) - } - ) - ) + stitched_choices: Final = [ # mutable-ok: choices is a List field; a tuple there breaks model_dump round-trips + rebuilt.choices[0].model_copy(update=MappingProxyType({"index": index})) + for index, rebuilt in rebuilt_by_index + ] + return base_response.model_copy(update=MappingProxyType({"choices": stitched_choices})) + + @staticmethod + def _narrowed_to_choice(response: "ModelResponseStream", index: int) -> "ModelResponseStream": + narrowed: Final = [choice for choice in response.choices if choice.index == index] # mutable-ok: List field + return response.model_copy(update=MappingProxyType({"choices": narrowed})) def build_stream_error_items( self, diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_moderations.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_moderations.py index 88b4ac7172a..c7adefe9886 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_moderations.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_moderations.py @@ -369,6 +369,7 @@ async def test_openai_moderation_guardrail_streaming_safe_content(): chunk1.choices[0].delta = MagicMock() chunk1.choices[0].delta.content = "Hello " chunk1.choices[0].finish_reason = None + chunk1.choices[0].index = 0 chunk2 = MagicMock() chunk2.model = "gpt-4" @@ -376,6 +377,7 @@ async def test_openai_moderation_guardrail_streaming_safe_content(): chunk2.choices[0].delta = MagicMock() chunk2.choices[0].delta.content = "world" chunk2.choices[0].finish_reason = None + chunk2.choices[0].index = 0 # Last chunk with finish_reason chunk3 = MagicMock() @@ -384,6 +386,7 @@ async def test_openai_moderation_guardrail_streaming_safe_content(): chunk3.choices[0].delta = MagicMock() chunk3.choices[0].delta.content = "!" chunk3.choices[0].finish_reason = "stop" + chunk3.choices[0].index = 0 for chunk in [chunk1, chunk2, chunk3]: yield chunk @@ -480,6 +483,7 @@ async def test_openai_moderation_guardrail_streaming_harmful_content(): chunk1.choices[0].delta = MagicMock() chunk1.choices[0].delta.content = "This is " chunk1.choices[0].finish_reason = None + chunk1.choices[0].index = 0 # Last chunk - with finish_reason to signal end of stream chunk2 = MagicMock() @@ -488,6 +492,7 @@ async def test_openai_moderation_guardrail_streaming_harmful_content(): chunk2.choices[0].delta = MagicMock() chunk2.choices[0].delta.content = "harmful content" chunk2.choices[0].finish_reason = "stop" + chunk2.choices[0].index = 0 for chunk in [chunk1, chunk2]: yield chunk diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_openai_moderation_streaming.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_openai_moderation_streaming.py index cb6772977ec..16f04073fae 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_openai_moderation_streaming.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/openai/test_openai_moderation_streaming.py @@ -42,6 +42,7 @@ async def test_openai_moderation_guardrail_streaming_latency(): choice.delta.content = content # Last chunk gets finish_reason choice.finish_reason = "stop" if i == len(chunks_data) - 1 else None + choice.index = 0 chunk.choices = [choice] yield chunk @@ -122,6 +123,7 @@ async def test_openai_moderation_guardrail_streaming_harmful_content(): choice.delta.content = content # Last chunk gets finish_reason choice.finish_reason = "stop" if i == len(chunks_data) - 1 else None + choice.index = 0 chunk.choices = [choice] yield chunk @@ -224,6 +226,7 @@ async def test_openai_moderation_streaming_end_of_stream_request_data_passthroug choice.delta = MagicMock() choice.delta.content = content choice.finish_reason = "stop" if i == len(chunks_data) - 1 else None + choice.index = 0 chunk.choices = [choice] yield chunk diff --git a/tests/test_litellm/proxy/utils/proxy_logging/test_guardrail_pipeline.py b/tests/test_litellm/proxy/utils/proxy_logging/test_guardrail_pipeline.py index 5f3c09d9195..8bd9dc0df8a 100644 --- a/tests/test_litellm/proxy/utils/proxy_logging/test_guardrail_pipeline.py +++ b/tests/test_litellm/proxy/utils/proxy_logging/test_guardrail_pipeline.py @@ -1836,6 +1836,22 @@ def _rewritten_model_response(response: Any) -> litellm.ModelResponse: return litellm.ModelResponse(**payload) +def _two_choice_stream_chunks() -> List[Any]: + return [ + litellm.ModelResponseStream(choices=[{"index": 0, "delta": {"content": "hello "}, "finish_reason": None}]), + litellm.ModelResponseStream(choices=[{"index": 1, "delta": {"content": "bonjour "}, "finish_reason": None}]), + litellm.ModelResponseStream(choices=[{"index": 0, "delta": {"content": "world"}, "finish_reason": "stop"}]), + litellm.ModelResponseStream(choices=[{"index": 1, "delta": {"content": "monde"}, "finish_reason": "stop"}]), + ] + + +def _rewritten_every_choice(response: Any) -> litellm.ModelResponse: + payload = response.model_dump() + for choice in payload["choices"]: + choice["message"]["content"] = "[REWRITTEN] " + choice["message"]["content"] + return litellm.ModelResponse(**payload) + + def test_streamable_post_call_pipelines_keeps_hook_guardrails_and_drops_iterator_only( make_user_api_key_auth, monkeypatch, caplog ): @@ -1984,6 +2000,39 @@ async def test_streaming_iterator_hook_runs_legacy_hook_and_delivers_its_rewrite assert _warnings(caplog) == [] +@pytest.mark.asyncio +async def test_streaming_iterator_hook_delivers_legacy_hook_rewrite_on_every_choice( + proxy_logging, make_user_api_key_auth, monkeypatch, caplog +): + seen: Dict[str, Any] = {} + guardrail = _legacy_hook_stream_guardrail(seen, rewrite=_rewritten_every_choice) + monkeypatch.setattr(litellm, "callbacks", [guardrail]) + monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None, raising=False) + data = _post_call_pipeline_data(stream=True) + chunks = _two_choice_stream_chunks() + auth = make_user_api_key_auth(request_route="/v1/chat/completions") + + with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"): + await proxy_logging.pre_call_hook(user_api_key_dict=auth, data=data, call_type="completion", guardrails_only=True) + delivered = [ + item + async for item in proxy_logging.async_post_call_streaming_iterator_hook( + user_api_key_dict=auth, response=_async_chunk_iter(chunks), request_data=data + ) + ] + + assert [choice.message.content for choice in seen["response"].choices] == ["hello world", "bonjour monde"] + assert [id(item) for item in delivered] == [id(chunk) for chunk in chunks] + assert [(item.choices[0].index, item.choices[0].delta.content) for item in delivered] == [ + (0, "[REWRITTEN] hello world"), + (1, "[REWRITTEN] bonjour monde"), + (0, ""), + (1, ""), + ] + assert [item.choices[0].finish_reason for item in delivered] == [None, None, "stop", "stop"] + assert _warnings(caplog) == [] + + @pytest.mark.asyncio async def test_streaming_iterator_hook_releases_stream_untouched_when_legacy_hook_returns_none( proxy_logging, make_user_api_key_auth, monkeypatch