fix(guardrails): withhold chat finish chunk in end_of_stream_only mode and close open Responses items before a mid-stream block

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]:

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,
+    )

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

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."

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