diff --git a/litellm/llms/openai/chat/guardrail_translation/handler.py b/litellm/llms/openai/chat/guardrail_translation/handler.py index 05716e65137..55b27947faa 100644 --- a/litellm/llms/openai/chat/guardrail_translation/handler.py +++ b/litellm/llms/openai/chat/guardrail_translation/handler.py @@ -1015,6 +1015,21 @@ class OpenAIChatCompletionsHandler(BaseTranslation): # Subsequent chunks - clear the text content_item["text"] = "" + def _check_streaming_has_ended(self, responses_so_far: Sequence[object]) -> bool: + """ + True once any relayed chunk carries a non-null ``finish_reason``. + + The unified guardrail's ``end_of_stream_only`` streaming path probes + this via ``hasattr`` to withhold the terminal chunks until + end-of-stream moderation runs, so a block can replace the finish + instead of trailing after a ``finish_reason`` the client already saw. + """ + return any( + stream_item_field(choice, "finish_reason") is not None + for item in responses_so_far + for choice in _stream_chunk_choices(item) + ) + def build_block_sse_chunks( self, exc: "ModifyResponseException", @@ -1097,6 +1112,13 @@ def _chat_sse_chunk(payload: _BlockedChunk) -> bytes: return f"data: {json.dumps(payload)}\n\n".encode() +def _stream_chunk_choices(item: object) -> Sequence[object]: + choices: Final = stream_item_field(item, "choices") + if isinstance(choices, Sequence) and not isinstance(choices, (str, bytes)): + return choices + return () + + def _blocked_stream_identity( exc: "ModifyResponseException", responses_so_far: Sequence[object] ) -> tuple[str, int, str]: diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py index 811c4feeacc..cf01242f151 100644 --- a/litellm/llms/openai/responses/guardrail_translation/handler.py +++ b/litellm/llms/openai/responses/guardrail_translation/handler.py @@ -31,6 +31,7 @@ Output: response.output is List[GenericResponseOutputItem] where each has: import time import uuid from collections.abc import Sequence +from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Final, Union, cast from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall @@ -874,10 +875,10 @@ class OpenAIResponsesHandler(BaseTranslation): sent, so emit the full synthetic sequence (``response.created`` through ``response.completed``). - ``stream_started`` True (sampling / mid-stream): events already - reached the client, so continue the in-progress response: deliver the - block message as a new output item under the same response id and - close with a ``response.completed`` carrying only the replacement - item. + reached the client, so continue the in-progress response: close the + output item still open on the wire, deliver the block message as a + new output item under the same response id, and close with a + ``response.completed`` carrying only the replacement item. The proxy's data generator appends ``data: [DONE]`` itself. """ @@ -887,7 +888,8 @@ class OpenAIResponsesHandler(BaseTranslation): else self._standalone_block_events(exc) ) return tuple( - f"data: {event.model_dump_json(exclude_none=True, exclude_unset=True)}\n\n".encode() for event in events + f"data: {event.model_dump_json(exclude_none=True, exclude_unset=True, serialize_as_any=True)}\n\n".encode() + for event in events ) @staticmethod @@ -915,6 +917,7 @@ class OpenAIResponsesHandler(BaseTranslation): "logprobs": None, } return ( + *_open_item_closing_events(responses_so_far), OutputItemAddedEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, output_index=output_index, @@ -1036,3 +1039,123 @@ def _continuation_identity(exc: "ModifyResponseException", responses_so_far: Seq index for item in responses_so_far if isinstance(index := stream_item_field(item, "output_index"), int) ) return response_id, model, max(indices) + 1 if indices else 0 + + +@dataclass(frozen=True, slots=True) +class _OpenItemState: + item_id: str + item_type: str + role: str + output_index: int + content_index: int + text: str + part_open: bool + + +def _open_item_state(responses_so_far: Sequence[object]) -> _OpenItemState | None: + typed: Final = tuple((str(stream_item_field(event, "type") or ""), event) for event in responses_so_far) + added: Final = tuple( + (added_index, stream_item_field(event, "item")) + for event_type, event in typed + if event_type == "response.output_item.added" + and isinstance(added_index := stream_item_field(event, "output_index"), int) + ) + done_indices: Final = frozenset( + done_index + for event_type, event in typed + if event_type == "response.output_item.done" + and isinstance(done_index := stream_item_field(event, "output_index"), int) + ) + open_added: Final = tuple((index, payload) for index, payload in added if index not in done_indices) + if not open_added: + return None + output_index, item_payload = open_added[-1] + item_id: Final = stream_item_field(item_payload, "id") if item_payload is not None else None + if not isinstance(item_id, str) or not item_id: + return None + raw_type: Final = stream_item_field(item_payload, "type") + raw_role: Final = stream_item_field(item_payload, "role") + part_added: Final = tuple( + part_index + for event_type, event in typed + if event_type == "response.content_part.added" + and stream_item_field(event, "item_id") == item_id + and isinstance(part_index := stream_item_field(event, "content_index"), int) + ) + part_done: Final = frozenset( + part_done_index + for event_type, event in typed + if event_type == "response.content_part.done" + and stream_item_field(event, "item_id") == item_id + and isinstance(part_done_index := stream_item_field(event, "content_index"), int) + ) + open_parts: Final = tuple(index for index in part_added if index not in part_done) + text: Final = "".join( + delta + for event_type, event in typed + if event_type == "response.output_text.delta" + and stream_item_field(event, "item_id") == item_id + and isinstance(delta := stream_item_field(event, "delta"), str) + ) + return _OpenItemState( + item_id=item_id, + item_type=raw_type if isinstance(raw_type, str) and raw_type else "message", + role=raw_role if isinstance(raw_role, str) and raw_role else "assistant", + output_index=output_index, + content_index=open_parts[-1] if open_parts else 0, + text=text, + part_open=bool(open_parts), + ) + + +def _open_item_closing_events(responses_so_far: Sequence[object]) -> Sequence[ResponsesAPIStreamingResponse]: + """Close the output item still in progress on the relayed stream before the + block item is appended: strict Responses clients reject a + ``response.completed`` that arrives while an earlier ``output_item.added`` + was never closed. The closing text is exactly what the client has received + for that item so far.""" + open_item: Final = _open_item_state(responses_so_far) + if open_item is None: + return () + partial_part: Final[_BlockedContentPart] = { + "type": "output_text", + "text": open_item.text, + "annotations": (), + } + closed_payload: Final[_BlockedItemPayload] = { + "type": open_item.item_type, + "id": open_item.item_id, + "status": "completed", + "role": open_item.role, + "content": (partial_part,), + } + item_done: Final = OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=open_item.output_index, + item=GenericResponseOutputItem.model_validate(closed_payload), + ) + if not open_item.part_open: + return (item_done,) + partial_done_part: Final[_BlockedDoneContentPart] = { + "type": "output_text", + "text": open_item.text, + "annotations": (), + "logprobs": None, + } + return ( + OutputTextDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + item_id=open_item.item_id, + output_index=open_item.output_index, + content_index=open_item.content_index, + text=open_item.text, + ), + ContentPartDoneEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_DONE, + item_id=open_item.item_id, + output_index=open_item.output_index, + content_index=open_item.content_index, + part=ContentPartDonePartOutputText.model_validate(partial_done_part), + ), + item_done, + ) 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 8ad0026d1ab..7dd6065063a 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 @@ -1610,3 +1610,36 @@ class TestBuildBlockSseChunks: assert first["choices"][0]["delta"] == {"content": "Blocked by policy."} assert final["id"] == "chatcmpl-live" assert final["usage"] == {"prompt_tokens": 11, "completion_tokens": 5, "total_tokens": 16} + + +class TestCheckStreamingHasEnded: + """_check_streaming_has_ended lets end_of_stream_only withhold the finish chunk until moderation""" + + def test_empty_and_content_only_chunks_are_not_ended(self): + handler = OpenAIChatCompletionsHandler() + assert handler._check_streaming_has_ended([]) is False + content_only = [ + {"id": "chatcmpl-live", "choices": [{"index": 0, "delta": {"content": "hi"}, "finish_reason": None}]}, + {"id": "chatcmpl-live", "choices": []}, + {"id": "chatcmpl-live", "usage": {"prompt_tokens": 1, "completion_tokens": 1}}, + ] + assert handler._check_streaming_has_ended(content_only) is False + + def test_dict_finish_chunk_marks_stream_ended(self): + handler = OpenAIChatCompletionsHandler() + chunks = [ + {"id": "chatcmpl-live", "choices": [{"index": 0, "delta": {"content": "hi"}, "finish_reason": None}]}, + {"id": "chatcmpl-live", "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}, + ] + assert handler._check_streaming_has_ended(chunks) is True + + def test_object_finish_chunk_marks_stream_ended(self): + from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices + + handler = OpenAIChatCompletionsHandler() + chunks = [ + ModelResponseStream( + choices=[StreamingChoices(index=0, delta=Delta(content=None), finish_reason="stop")] + ) + ] + assert handler._check_streaming_has_ended(chunks) is True 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 cd0e7f29933..1ae4e7a699b 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 @@ -1285,10 +1285,32 @@ class TestBuildBlockSseChunks: ) types = [payload["type"] for payload in payloads] assert "response.created" not in types - assert types[0] == "response.output_item.added" - assert payloads[0]["output_index"] == 3 + assert types[0] == "response.output_item.done" + assert payloads[0]["output_index"] == 2 + assert payloads[0]["item"]["id"] == "msg_orig" + assert payloads[0]["item"]["status"] == "completed" + assert types[1] == "response.output_item.added" + assert payloads[1]["output_index"] == 3 completed = payloads[-1]["response"] assert completed["id"] == "resp_live" assert completed["model"] == "gpt-5.4-mini-2026-01-01" assert completed["output"][0]["content"][0]["text"] == "Blocked by policy." assert completed["usage"] == {"input_tokens": 7, "output_tokens": 21, "total_tokens": 28} + + def test_continuation_without_open_item_emits_no_closing_events(self): + handler = OpenAIResponsesHandler() + yielded = [ + {"type": "response.created", "response": {"id": "resp_live", "model": "gpt-5.4-mini"}}, + {"type": "response.in_progress", "response": {"id": "resp_live"}}, + ] + payloads = self._payloads( + handler.build_block_sse_chunks( + self._exc(original_response=yielded), stream_started=True, responses_so_far=yielded + ) + ) + types = [payload["type"] for payload in payloads] + assert types[0] == "response.output_item.added" + assert types[-1] == "response.completed" + dones = [payload for payload in payloads if payload["type"] == "response.output_item.done"] + assert len(dones) == 1 + assert dones[0]["item"]["content"][0]["text"] == "Blocked by policy." diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py index 99172ec3be6..42bdf41bc88 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py @@ -53,6 +53,19 @@ class _BlockingGuardrail(CustomGuardrail): ) +class _PassingGuardrail(CustomGuardrail): + """Mock guardrail that always lets response scans through unchanged.""" + + async def apply_guardrail( + self, + inputs: GenericGuardrailAPIInputs, + request_data: dict, + input_type: Literal["request", "response"], + logging_obj: Optional[Any] = None, + ) -> GenericGuardrailAPIInputs: + return inputs + + def _chat_chunk(delta: Delta, finish_reason: Optional[str] = None) -> ModelResponseStream: return ModelResponseStream( id="chatcmpl-live", @@ -129,8 +142,13 @@ async def _run_hook( sampling_rate: int = 1, end_of_stream_only: bool = False, buffer_until_moderated: bool = False, + blocks: bool = True, ) -> Tuple[StreamChunk, ...]: - guardrail = _BlockingGuardrail(guardrail_name="test-blocking-guardrail", event_hook="post_call") + guardrail = ( + _BlockingGuardrail(guardrail_name="test-blocking-guardrail", event_hook="post_call") + if blocks + else _PassingGuardrail(guardrail_name="test-passing-guardrail", event_hook="post_call") + ) guardrail.streaming_sampling_rate = sampling_rate guardrail.streaming_end_of_stream_only = end_of_stream_only guardrail.streaming_buffer_until_moderated = buffer_until_moderated @@ -140,7 +158,7 @@ async def _run_hook( request_data = { "messages": [{"role": "user", "content": "hi"}], "guardrail_to_apply": guardrail, - "metadata": {"guardrails": ["test-blocking-guardrail"]}, + "metadata": {"guardrails": [guardrail.guardrail_name]}, } return tuple( @@ -203,13 +221,38 @@ async def test_chat_mid_stream_block_continues_the_completion(): @pytest.mark.asyncio async def test_chat_end_of_stream_block_terminates_cleanly(): + """Regression for bugbot's finish-ordering finding: in end_of_stream_only + mode the original finish chunk must be withheld until moderation decides, + so a block's content_filter finish is the only stream terminator a client + ever sees - never policy text trailing after finish_reason stop.""" collected = await _run_hook("/v1/chat/completions", _chat_stream(end=True), end_of_stream_only=True) _assert_no_error_frame(collected) + forwarded = [chunk for chunk in collected if isinstance(chunk, ModelResponseStream)] + assert forwarded, "content chunks still stream to the client before end-of-stream moderation" + assert all(choice.finish_reason is None for chunk in forwarded for choice in chunk.choices), ( + "the original finish chunk must be withheld until moderation decides" + ) payloads = _sse_payloads(collected) assert BLOCK_MESSAGE in json.dumps(payloads) assert payloads[-1]["choices"][0]["finish_reason"] == "content_filter" +@pytest.mark.asyncio +async def test_chat_end_of_stream_pass_releases_withheld_finish_chunk(): + """When end-of-stream moderation passes, the withheld finish chunk is + released so a clean stream still terminates normally.""" + collected = await _run_hook( + "/v1/chat/completions", _chat_stream(end=True), end_of_stream_only=True, blocks=False + ) + assert not [chunk for chunk in collected if isinstance(chunk, bytes)], ( + "a clean stream must carry no synthetic block frames" + ) + forwarded = [chunk for chunk in collected if isinstance(chunk, ModelResponseStream)] + finish_reasons = [choice.finish_reason for chunk in forwarded for choice in chunk.choices] + assert finish_reasons[-1] == "stop", "the withheld finish chunk must be released after moderation passes" + assert all(reason is None for reason in finish_reasons[:-1]) + + @pytest.mark.asyncio async def test_responses_buffered_block_emits_full_event_sequence(): """Buffered moderation blocks before anything streams: a complete synthetic @@ -234,20 +277,35 @@ async def test_responses_buffered_block_emits_full_event_sequence(): @pytest.mark.asyncio async def test_responses_mid_stream_block_continues_the_response(): - """Regression for the LIT-6496 500 error frame: after events were already - forwarded, the block appends a new output item under the same response id - and closes with response.completed - never a second response.created.""" + """Regression for the LIT-6496 500 error frame and bugbot's unclosed-item + finding: after events were already forwarded, the block first closes the + output item still open on the wire, then appends the replacement item under + the same response id, and closes with response.completed - never a second + response.created and never a completed response with an item left open.""" collected = await _run_hook("/v1/responses", _responses_stream(end=False)) _assert_no_error_frame(collected) - forwarded_types = [chunk["type"] for chunk in collected if isinstance(chunk, dict)] + forwarded = [chunk for chunk in collected if isinstance(chunk, dict)] + forwarded_types = [chunk["type"] for chunk in forwarded] assert "response.created" in forwarded_types, "original events should have streamed before the block" payloads = _sse_payloads(collected) assert payloads, "no block SSE chunks were emitted" block_types = [payload["type"] for payload in payloads] assert "response.created" not in block_types, "a mid-stream block must not restart the response" - assert block_types[0] == "response.output_item.added" assert block_types[-1] == "response.completed" - assert payloads[0]["output_index"] == 1, "the block item must continue after the original output item" + + all_events = forwarded + list(payloads) + opened = sorted(event["output_index"] for event in all_events if event["type"] == "response.output_item.added") + closed = sorted(event["output_index"] for event in all_events if event["type"] == "response.output_item.done") + assert opened == closed, "every output item opened on the stream must be closed before response.completed" + original_done_position = block_types.index("response.output_item.done") + block_item_position = block_types.index("response.output_item.added") + assert original_done_position < block_item_position, ( + "the in-progress original item must be closed before the block item is appended" + ) + assert payloads[original_done_position]["item"]["id"] == "msg_orig" + assert payloads[block_item_position]["output_index"] == 1, ( + "the block item must continue after the original output item" + ) completed = payloads[-1]["response"] assert completed["id"] == "resp_live" assert completed["output"][0]["content"][0]["text"] == BLOCK_MESSAGE