fix(guardrails): deliver modify_response block as valid SSE on streaming chat and Responses

[mateo-berri]

TLDR

Problem this solves:

• Streaming guardrail replacements break the stream on chat and Responses
• Mid-stream blocks arrive as an SSE error frame, no [DONE]
• Buffered streams return a plain HTTP 500 body, no SSE at all
• Only /v1/messages already delivered blocks as a valid stream

How it solves it:

• Chat handler emits the replacement delta, then a content_filter finish chunk
• Responses handler emits typed events ending in response.completed
• Pre-stream (buffered) blocks send a full synthetic stream
• In end-of-stream-only mode, chat withholds the finish chunk until moderation runs, so a block never trails an already-sent finish_reason
• When a block lands while a Responses output item is still open, the handler closes that item first, then appends the replacement at the next output_index
• A non-message item still open at the block (a mid-stream tool call) is closed with status: "incomplete" and its fields preserved, so the synthetic done event never authorizes acting on it
• Both close cleanly with data: [DONE] and HTTP 200

User Flow

Before: a developer streaming through a gateway whose guardrail replaces policy-violating answers gets a broken stream on chat and Responses, and an outright 500 when the guardrail buffers

1. They send POST http://localhost:4000/v1/chat/completions with "stream": true and a prompt whose answer trips the guardrail
2. The disallowed words stream down first, then the stream aborts with data: {"error": {"message": "This response was replaced by policy.", "code": "500"}} and data: [DONE] never arrives, so their SDK raises mid-stream
3. The same request to POST http://localhost:4000/v1/responses streams the normal response events, then dies on the same error frame with no [DONE]
4. When the guardrail's stream buffering is turned on to stop the leak, the same two requests return HTTP 500 with a bare JSON error body and no stream at all
5. With a guardrail that only moderates at end of stream (streaming_end_of_stream_only = True), chat relays the whole answer including its finish_reason: "stop" chunk, then tacks the same error frame onto an already-finished stream
6. Only POST http://localhost:4000/v1/messages streams the replacement text and ends cleanly with message_stop

After: the same requests deliver the replacement as a well-formed stream that ends cleanly

1. They send the same POST http://localhost:4000/v1/chat/completions with "stream": true
2. The stream ends with a chunk carrying "This response was replaced by policy.", a finish_reason: "content_filter" chunk with real token usage, then data: [DONE], all under HTTP 200
3. The same request to POST http://localhost:4000/v1/responses delivers the replacement as a new output item and closes with response.completed and data: [DONE], HTTP 200; if the block lands while a message item is mid-flight, that item is first closed with output_text.done, content_part.done, and output_item.done carrying the text relayed so far, and a mid-flight non-message item (a streaming tool call) is instead closed by an output_item.done marking it "incomplete", then the replacement item follows at the next output_index
4. With stream buffering on, nothing from the original answer leaks: chat sends only the replacement chunks, and Responses sends a complete response.created through response.completed sequence, both HTTP 200
5. With the end-of-stream-only guardrail, chat withholds the finish chunk while moderation runs: a block delivers the replacement delta and a content_filter finish, and a clean answer gets its withheld stop finish released unchanged
6. POST http://localhost:4000/v1/messages behaves the same as before

Relevant issues

Linear ticket

Resolves LIT-6496

Pre-Submission checklist

Please complete all items before asking a LiteLLM maintainer to review your PR

• I have added meaningful tests
• The handful of test files covering my change pass locally, e.g. uv run pytest tests/test_litellm/<your_test_file>.py -v. Leave the suites (make test-unit-*, make test-unit) to CI: it finishes in ~15 minutes where a laptop takes an hour or more
• My PR passes all required CI/CD checks (e.g., lint, schema.d.ts sync check, etc.)
• My PR's scope is as isolated as possible; it only solves 1 specific problem
• I have received a Greptile Confidence Score of at least 4/5 before requesting a maintainer review (Greptile reviews automatically once the PR is opened; only comment @greptileai to re-request a review after pushing changes)

Delays in PR merge?

If you're seeing a delay in your PR being merged, ping the LiteLLM Team on Slack (#pr-review).

Screenshots / Proof of Fix

QA ran at 4567fc7 against its merge base 4517e5e, superseding the earlier run at 38825cf. The tip has since moved twice: a38dfec is a staging merge that leaves the PR diff byte-identical, and 8c7fe00 fixes the usage lookup in blocked_responses_stream_usage (Bugbot's catch: the old code compared str(event.type) against "response.completed", which never matches the enum's class-qualified str). The run below still proves the block and close flows; the one behavior it showed that has since changed is the buffered Responses usage, which the run reported as zero (see s4). That leg was re-run live at 8c7fe00 (step 4 under the buffered section) and now carries the real upstream usage, with regression test test_continuation_reads_usage_from_typed_completed_event pinning it

Shared setup: a live proxy per side against real OpenAI and Anthropic APIs, with a custom post_call guardrail that replaces any answer containing "kumquat"

lit6496_guardrail.py:

from typing import Any, Literal, Optional

from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.types.utils import GenericGuardrailAPIInputs


class ReplaceOnKumquat(CustomGuardrail):
    async def apply_guardrail(
        self,
        inputs: GenericGuardrailAPIInputs,
        request_data: dict,
        input_type: Literal["request", "response"],
        logging_obj: Optional[Any] = None,
    ) -> GenericGuardrailAPIInputs:
        if input_type != "response":
            return inputs
        texts = inputs.get("texts") or []
        if any("kumquat" in text.lower() for text in texts if isinstance(text, str)):
            self.raise_passthrough_exception(
                violation_message="This response was replaced by policy.",
                request_data=request_data,
            )
        return inputs

lit6496_config.yaml (two variant configs point at copies of the same class: one with streaming_buffer_until_moderated = True set on it for the buffered legs, one with streaming_end_of_stream_only = True for the end-of-stream legs):

model_list:
  - model_name: gpt-5.4-mini
    litellm_params:
      model: openai/gpt-5.4-mini
      api_key: os.environ/OPENAI_API_KEY
  - model_name: claude-sonnet-5
    litellm_params:
      model: anthropic/claude-sonnet-5
      api_key: os.environ/ANTHROPIC_API_KEY

guardrails:
  - guardrail_name: policy-replacer
    litellm_params:
      guardrail: lit6496_guardrail.ReplaceOnKumquat
      mode: post_call
      default_on: true

general_settings:
  master_key: sk-lit6496-qa

Boot per side (base at the merge base, after at this PR's tip), one proxy per config, each with 2 uvicorn workers; the streaming block path holds no cross-process state, so both legs run this same topology:

python litellm/proxy/proxy_cli.py --config lit6496_config.yaml --port <port> --num_workers 2

Request bodies used throughout (Authorization: Bearer sk-lit6496-qa):

PROMPT='Reply with exactly the single word kumquat, lowercase, nothing else.'
curl -sS -w '\nHTTP_STATUS:%{http_code}\n' http://localhost:<port>/v1/chat/completions -H "Authorization: Bearer sk-lit6496-qa" -H 'Content-Type: application/json' -d "{\"model\":\"gpt-5.4-mini\",\"stream\":true,\"messages\":[{\"role\":\"user\",\"content\":\"$PROMPT\"}]}"
curl -sS -w '\nHTTP_STATUS:%{http_code}\n' http://localhost:<port>/v1/responses -H "Authorization: Bearer sk-lit6496-qa" -H 'Content-Type: application/json' -d "{\"model\":\"gpt-5.4-mini\",\"stream\":true,\"input\":\"$PROMPT\"}"
curl -sS -w '\nHTTP_STATUS:%{http_code}\n' http://localhost:<port>/v1/messages -H "Authorization: Bearer sk-lit6496-qa" -H 'Content-Type: application/json' -d "{\"model\":\"claude-sonnet-5\",\"max_tokens\":100,\"stream\":true,\"messages\":[{\"role\":\"user\",\"content\":\"$PROMPT\"}]}"

Before (4517e5e)

/v1/chat/completions streaming

1. Send the chat curl above against the base proxy
2. The stream leaks the original answer, then dies on an error frame with no [DONE]:

data: {"id":"chatcmpl-EJPqnT3jZ97Ly3rnIFvUYazQDS3nd","object":"chat.completion.chunk","created":1788296309,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"kum"}}]}

data: {"id":"chatcmpl-EJPqnT3jZ97Ly3rnIFvUYazQDS3nd","object":"chat.completion.chunk","created":1788296309,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"quat"}}]}

data: {"id":"chatcmpl-EJPqnT3jZ97Ly3rnIFvUYazQDS3nd","object":"chat.completion.chunk","created":1788296309,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: {"error": {"message": "This response was replaced by policy.", "type": "None", "param": "None", "code": "500"}}


HTTP_STATUS:200

/v1/responses streaming

1. Send the Responses curl above against the base proxy
2. The normal response events stream down, then the stream ends on the same error frame with no [DONE]:

data: {"type":"response.output_item.done","output_index":0,"sequence_number":8,"item":{"id":"msg_09cbbab542100cd5006a973c76a05887d0ac60a514a0d21fdb","type":"message","status":"completed","content":[{"type":"output_text","annotations":[],"logprobs":[],"text":"kumquat"}],"phase":"final_answer","role":"assistant"},"model":"gpt-5.4-mini"}

data: {"error": {"message": "This response was replaced by policy.", "type": "None", "param": "None", "code": "500"}}


HTTP_STATUS:200

/v1/messages streaming

1. Send the messages curl above against the base proxy
2. This route already worked: the original words leak and the upstream stream closes, then a new content block carries the replacement and closes the stream again:

event: content_block_delta
data: {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "This response was replaced by policy."}}

event: content_block_stop
data: {"type": "content_block_stop", "index": 1}

event: message_delta
data: {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": null}, "usage": {"output_tokens": 6}}

event: message_stop
data: {"type": "message_stop"}


HTTP_STATUS:200

buffered guardrail, /v1/chat/completions and /v1/responses streaming

1. Send the chat and Responses curls against the base proxy running the buffered guardrail
2. Both return the raw 500 with no SSE at all (the exact symptom from the ticket):

{"error":{"message":"This response was replaced by policy.","type":"None","param":"None","code":"500"}}
HTTP_STATUS:500

end-of-stream-only guardrail, /v1/chat/completions streaming

1. Send the chat curl against the base proxy running the end-of-stream-only guardrail
2. The whole answer streams down including the finish_reason: "stop" chunk, then the error frame lands on an already-finished stream:

data: {"id":"chatcmpl-EJPsUP6bDg2cQOqhCle6tZch7F6Yn","object":"chat.completion.chunk","created":1788296414,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"kum"}}]}

data: {"id":"chatcmpl-EJPsUP6bDg2cQOqhCle6tZch7F6Yn","object":"chat.completion.chunk","created":1788296414,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"quat"}}]}

data: {"id":"chatcmpl-EJPsUP6bDg2cQOqhCle6tZch7F6Yn","object":"chat.completion.chunk","created":1788296414,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: {"error": {"message": "This response was replaced by policy.", "type": "None", "param": "None", "code": "500"}}


HTTP_STATUS:200

streamed tool call, /v1/responses

1. Send a Responses request with a record_word function tool and a prompt forcing a call with the word kumquat
2. The post_call streaming guardrail never fires: texts carries only output text, so function_call arguments are not moderated. The call streams through untouched, closes status: "completed", and the stream ends normally with response.completed and [DONE], HTTP 200. The defect on this path is non-detection, not a broken stream; detection is out of scope here (see caveats)

non-streaming, all three routes

1. Send the three curls without "stream": true against the base proxy
2. All three return HTTP 200 with the replacement (chat shown; unchanged by this PR):

{"id":"chatcmpl-73e6fb1f-aa87-4ce7-898c-ea4b09b52d48","created":1788296318,"model":"gpt-5.4-mini","object":"chat.completion","system_fingerprint":null,"choices":[{"finish_reason":"content_filter","index":0,"message":{"content":"This response was replaced by policy.","role":"assistant","tool_calls":null,"function_call":null,"provider_specific_fields":null}}],"usage":{...}}
HTTP_STATUS:200

After (4567fc7)

/v1/chat/completions streaming

1. Send the chat curl above against the tip proxy
2. The sampled stream still relays pre-block chunks (pre-existing, see caveats), but now ends with the replacement delta, a content_filter finish chunk carrying real usage, and [DONE]:

data: {"id":"chatcmpl-EJQKL9AU60IZKxjhBa59QGT6HOLO5","object":"chat.completion.chunk","created":1788298141,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"quat"}}]}

data: {"id":"chatcmpl-EJQKL9AU60IZKxjhBa59QGT6HOLO5","object":"chat.completion.chunk","created":1788298141,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: {"id": "chatcmpl-EJQKL9AU60IZKxjhBa59QGT6HOLO5", "object": "chat.completion.chunk", "created": 1788298141, "model": "gpt-5.4-mini", "choices": [{"index": 0, "delta": {"content": "This response was replaced by policy."}, "finish_reason": null}]}

data: {"id": "chatcmpl-EJQKL9AU60IZKxjhBa59QGT6HOLO5", "object": "chat.completion.chunk", "created": 1788298141, "model": "gpt-5.4-mini", "choices": [{"index": 0, "delta": {}, "finish_reason": "content_filter"}], "usage": {"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25}}

data: [DONE]


HTTP_STATUS:200

/v1/responses streaming

1. Send the Responses curl above against the tip proxy
2. The replacement arrives as a new output item under the live response id and the stream closes with response.completed and [DONE]:

data: {"type":"response.output_text.delta","item_id":"msg_e1c8d690-e458-4da5-9efb-6695f903487f","output_index":1,"content_index":0,"delta":"This response was replaced by policy."}

data: {"type":"response.output_item.done","output_index":1,"item":{"type":"message","id":"msg_e1c8d690-e458-4da5-9efb-6695f903487f","status":"completed","role":"assistant","content":[{"type":"output_text","text":"This response was replaced by policy.","annotations":[]}]}}

data: {"type":"response.completed","response":{"id":"resp_...","created_at":1788298142,"model":"gpt-5.4-mini-2026-03-17","object":"response","output":[{"type":"message","id":"msg_e1c8d690-e458-4da5-9efb-6695f903487f","status":"completed","role":"assistant","content":[{"type":"output_text","text":"This response was replaced by policy.","annotations":[]}]}],"status":"completed","usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}

data: [DONE]


HTTP_STATUS:200

mid-answer block on /v1/responses: the open message item gets closed first

1. Send a longer prompt so the block lands while output item 0 is still open (its output_text.done has not arrived yet):

curl -sS -w '\nHTTP_STATUS:%{http_code}\n' http://localhost:<port>/v1/responses -H "Authorization: Bearer sk-lit6496-qa" -H 'Content-Type: application/json' -d '{"model":"gpt-5.4-mini","stream":true,"input":"Write a 150 word essay about the kumquat fruit."}'

2. The upstream deltas relay, then the proxy closes item 0 with a synthesized output_text.done / content_part.done / output_item.done trio carrying exactly the text relayed so far (synthesized frames carry no sequence_number), then the replacement item follows at output_index 1 and the stream completes with a single response.completed, HTTP 200:

data: {"type":"response.output_text.done","item_id":"msg_02a0a8e12031b210006a974419ae5c87d0956d236599a2a36d","output_index":0,"content_index":0,"text":"The kumquat is a small, bright citrus fruit that packs a surprising amount of flavor..."}

data: {"type":"response.content_part.done","item_id":"msg_02a0a8e12031b210006a974419ae5c87d0956d236599a2a36d","output_index":0,"content_index":0,"part":{"type":"output_text","text":"The kumquat is a small, bright citrus fruit...","annotations":[]}}

data: {"type":"response.output_item.done","output_index":0,"item":{"type":"message","id":"msg_02a0a8e12031b210006a974419ae5c87d0956d236599a2a36d","status":"completed","role":"assistant","content":[{"type":"output_text","text":"The kumquat is a small, bright citrus fruit...","annotations":[]}]}}

data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","id":"msg_aa4ab3fc-e0cf-4a09-a23c-653b5189d14b","status":"completed","role":"assistant","content":[{"type":"output_text","text":"This response was replaced by policy.","annotations":[]}]}}
...
data: {"type":"response.completed","response":{"id":"resp_...","status":"completed",...}}

data: [DONE]


HTTP_STATUS:200

When the async moderation scan instead loses the race outright and the block lands after the upstream stream already finished, the client sees the upstream close frames and response.completed, then the replacement item and a second response.completed, still ending [DONE] with HTTP 200 (see caveats)

blocked mid-stream tool call on /v1/responses: the open function_call closes incomplete

1. Send a prompt that produces kumquat text followed by a tool call, so the block lands while the function_call item is open (its arguments still streaming):

curl -sS -w '\nHTTP_STATUS:%{http_code}\n' http://localhost:<port>/v1/responses -H "Authorization: Bearer sk-lit6496-qa" -H 'Content-Type: application/json' -d '{"model":"gpt-5.4-mini","stream":true,"input":"Write three sentences about the kumquat fruit, then call the record_word tool with the word kumquat. You must call the tool.","tools":[{"type":"function","name":"record_word","description":"Record a word","parameters":{"type":"object","properties":{"word":{"type":"string"}},"required":["word"]}}]}'

2. The open function_call item is closed by a synthetic output_item.done marking it "incomplete", preserving its id, call_id, name, and the arguments relayed so far, fabricating no content; the replacement message item follows at the next output_index and the stream completes, HTTP 200:

data: {"type":"response.output_item.added","output_index":1,"item":{"id":"fc_0bffe7fe82973762006a974435874487d08314cefcbec96dbe","type":"function_call","status":"in_progress","arguments":"","call_id":"call_IeunFuqW0CjUDiYxkvvntrDm","name":"record_word"},"sequence_number":73,"model":"gpt-5.4-mini"}

data: {"type":"response.output_item.done","output_index":1,"item":{"id":"fc_0bffe7fe82973762006a974435874487d08314cefcbec96dbe","type":"function_call","status":"incomplete","arguments":"","call_id":"call_IeunFuqW0CjUDiYxkvvntrDm","name":"record_word"}}

data: {"type":"response.output_item.added","output_index":2,"item":{"type":"message","id":"msg_306664ab-abfe-483a-9ea0-1b0c755ba151","status":"completed","role":"assistant","content":[{"type":"output_text","text":"This response was replaced by policy.","annotations":[]}]}}
...
data: {"type":"response.completed",...}

data: [DONE]


HTTP_STATUS:200

A pure tool-call stream (no text output) still never trips the guardrail at the tip, same as before the PR: function_call arguments are not part of the moderated texts. That detection gap predates this PR and is out of scope (see caveats)

/v1/messages streaming

1. Send the messages curl above against the tip proxy
2. Unchanged: replacement content block, then a clean message_stop, HTTP 200

buffered guardrail, /v1/chat/completions and /v1/responses streaming

1. Send the chat and Responses curls against the tip proxy running the buffered guardrail
2. Chat returns only the replacement chunks with zero leaked content, then [DONE], HTTP 200:

data: {"id": "chatcmpl-25db5dd9-a171-4826-a64e-9e4f26298f94", "object": "chat.completion.chunk", "created": 1788298329, "model": "gpt-5.4-mini", "choices": [{"index": 0, "delta": {"role": "assistant", "content": "This response was replaced by policy."}, "finish_reason": null}]}

data: {"id": "chatcmpl-25db5dd9-a171-4826-a64e-9e4f26298f94", "object": "chat.completion.chunk", "created": 1788298329, "model": "gpt-5.4-mini", "choices": [{"index": 0, "delta": {}, "finish_reason": "content_filter"}], "usage": {"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25}}

data: [DONE]


HTTP_STATUS:200

3. Responses returns the full synthetic sequence with zero leaked content ("kumquat" appears nowhere in the transcript), HTTP 200:

response.created -> response.in_progress -> response.output_item.added -> response.content_part.added -> response.output_text.delta -> response.output_text.done -> response.content_part.done -> response.output_item.done -> response.completed

data: [DONE]


HTTP_STATUS:200

4. Re-run of this Responses leg at 8c7fe00 (same config, live OpenAI, 2 workers) to confirm the usage fix: the continuation's response.completed now carries the real upstream usage instead of zeros:

data: {"type":"response.completed","response":{"id":"resp_e670dab5-7559-4161-bd98-669a90a67ecd","created_at":1788309970,"model":"gpt-5.4-mini","object":"response","output":[{"type":"message","id":"msg_b3f0ba36-b5c2-4fd7-8294-5a257c812750","status":"completed","role":"assistant","content":[{"type":"output_text","text":"This response was replaced by policy.","annotations":[]}]}],"status":"completed","usage":{"input_tokens":20,"input_tokens_details":{"cached_tokens":0,"cache_write_tokens":0},"output_tokens":6,"output_tokens_details":{"reasoning_tokens":0},"total_tokens":26}}}

data: [DONE]


HTTP_STATUS:200

end-of-stream-only guardrail, /v1/chat/completions streaming

1. Send the chat curl against the tip proxy running the end-of-stream-only guardrail
2. The finish chunk is withheld while moderation runs; on a block the client gets the replacement delta and a content_filter finish instead, with no stray stop finish before it, then [DONE], HTTP 200:

data: {"id":"chatcmpl-EJQNtVul27AybkXm6dNmCFXIvcPFl","object":"chat.completion.chunk","created":1788298362,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"kum"}}]}

data: {"id":"chatcmpl-EJQNtVul27AybkXm6dNmCFXIvcPFl","object":"chat.completion.chunk","created":1788298362,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"quat"}}]}

data: {"id": "chatcmpl-EJQNtVul27AybkXm6dNmCFXIvcPFl", "object": "chat.completion.chunk", "created": 1788298362, "model": "gpt-5.4-mini", "choices": [{"index": 0, "delta": {"content": "This response was replaced by policy."}, "finish_reason": null}]}

data: {"id": "chatcmpl-EJQNtVul27AybkXm6dNmCFXIvcPFl", "object": "chat.completion.chunk", "created": 1788298362, "model": "gpt-5.4-mini", "choices": [{"index": 0, "delta": {}, "finish_reason": "content_filter"}], "usage": {"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25}}

data: [DONE]


HTTP_STATUS:200

3. On a clean answer (prompt asks for "hello") the withheld stop finish is released unchanged:

data: {"id":"chatcmpl-EJQNuwfEXIhi8qQ9nPdYoYWUpG7um","object":"chat.completion.chunk","created":1788298362,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{"content":"hello"}}]}

data: {"id":"chatcmpl-EJQNuwfEXIhi8qQ9nPdYoYWUpG7um","object":"chat.completion.chunk","created":1788298362,"model":"gpt-5.4-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]


HTTP_STATUS:200

non-streaming, all three routes

1. Send the three curls without "stream": true against the tip proxy
2. All three still return HTTP 200 with the replacement, byte-identical shape to before (chat shown):

{"id":"chatcmpl-e35589f9-e1fe-4144-a339-f86ff9cf980e","created":1788298148,"model":"gpt-5.4-mini","object":"chat.completion","system_fingerprint":null,"choices":[{"finish_reason":"content_filter","index":0,"message":{"content":"This response was replaced by policy.","role":"assistant","tool_calls":null,"function_call":null,"provider_specific_fields":null}}],"usage":{...}}
HTTP_STATUS:200

Observations from the run:

• Non-buffered streaming leaks pre-block chunks; pre-existing, unchanged here
• When async moderation loses the race entirely, the stream carries the upstream response.completed plus the continuation's second one; adjacent to the leak caveat
• Responses continuation response.completed reported zero usage at the QA commit; fixed at 8c7fe00 (equality instead of str() on the event type), covered by a regression test and re-proven live in step 4 of the buffered section
• Function call arguments are not part of the moderated texts, so a pure tool-call stream is never blocked; pre-existing detection gap
• /v1/messages appends the replacement after the upstream message_stop, so that stream carries two message_stop events; route untouched by this PR
• Guardrail subclasses inheriting apply_guardrail skip streaming moderation; pre-existing

Type

🐛 Bug Fix

Caveats (if any)

Low

• Non-buffered streaming still leaks pre-moderation chunks before the replacement; that sampling behavior predates this PR, and buffering already covers it. In the losing-race case the leak includes the upstream response.completed (chat: the upstream stop finish), so the fixed stream then carries a second one; a strict client may finalize at the first. Buffered and end-of-stream-only modes have no such race
• The Responses continuation's response.completed reports zero usage only when no upstream response.completed arrived before the block, so no usage exists to carry; buffered and end-of-stream blocks report the real upstream usage as of 8c7fe00
• Found while testing, out of scope here: function_call arguments are not included in the texts handed to streaming moderation, so a pure tool-call stream is never inspected on any route; pre-existing detection gap, tracked separately
• Found while testing, out of scope here: /v1/messages delivers mid-stream replacements after the upstream message_stop, so a strict Anthropic client may stop reading before the replacement block; pre-existing, route untouched by this PR
• Found while testing, out of scope here: a guardrail subclass that inherits apply_guardrail is skipped by streaming moderation entirely (leaf-class capability check); tracked separately
• A client that detected streaming blocks by catching the old in-band error frame now needs to key off finish_reason: content_filter (Responses: the replacement output item), the same signal the non-streaming path already sends

Final Attestation

• The tests check the right things, including the edge cases, and regressions in the respective real-world customer use-cases are not possible after this PR

• 3296547 passes /live-pr-risk

• a38dfec passes /live-pr-risk

• 8c7fe00 passes /live-pr-risk

[codspeed]

Merging this PR will not alter performance

✅ 31 untouched benchmarks

---

_{Comparing litellm_fix_stream_modify_response_chunks (8c7fe00) with litellm_internal_staging (3dac3f7)}

[greptile-apps]

Greptile Summary

The PR changes guardrail replacement handling so streaming Chat Completions and Responses requests terminate with valid endpoint-specific SSE

• Adds synthetic block chunks and usage handling for Chat Completions
• Adds complete and continuation event sequences for the Responses API, including open-item closure
• Withholds terminal events during end-of-stream-only moderation and expands regression coverage

Confidence Score: 5/5

The PR appears safe to merge

No blocking failure remains

Important Files Changed

Filename	Overview
litellm/llms/openai/chat/guardrail_translation/handler.py	Adds Chat Completions block SSE construction, stream identity reuse, usage reporting, and finish-chunk detection
litellm/llms/openai/responses/guardrail_translation/handler.py	Adds standalone and continuation block event construction with explicit closure of open message and non-message items
litellm/llms/base_llm/guardrail_translation/utils.py	Adds shared field access and blocked-stream usage extraction for chat and Responses event shapes
litellm/responses/streaming_iterator.py	Exposes synthetic Responses event generation for reuse by guardrail translation
tests/test_litellm/proxy/guardrails/guardrail_hooks/unified_guardrails/test_openai_streaming_block.py	Adds regression coverage for buffered, mid-stream, and end-of-stream block delivery across both OpenAI streaming APIs

_{Reviews (7): Last reviewed commit: "fix: compare stream event types by equal..." | Re-trigger Greptile}

[codecov]

Codecov Report

❌ Patch coverage is 98.25581% with 3 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
.../openai/responses/guardrail_translation/handler.py	97.19%	3 Missing ⚠️

📢 Thoughts on this report? Let us know!

[mateo-berri]

@greptileai

[mateo-berri]

bugbot run

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 2 potential issues.

Autofix Details

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Chat block follows already-sent finish
  • Added _check_streaming_has_ended to the OpenAI chat handler so the unified streaming hook withholds the terminal finish_reason chunk in end_of_stream_only mode, letting the block replacement replace (rather than follow) the finish.
• ✅ Fixed: Continuation leaves prior output item open
  • The Responses continuation now synthesizes output_text.done, content_part.done, and output_item.done events for any output item that was added but not closed before the block, so the replacement item and response.completed no longer arrive with a still-open earlier item.

Or push these changes by commenting:

@cursor push 0bad2bf7ed

Preview (0bad2bf7ed)

diff --git a/litellm/llms/openai/chat/guardrail_translation/handler.py b/litellm/llms/openai/chat/guardrail_translation/handler.py
--- a/litellm/llms/openai/chat/guardrail_translation/handler.py
+++ b/litellm/llms/openai/chat/guardrail_translation/handler.py
@@ -1015,6 +1015,23 @@
                                     # Subsequent chunks - clear the text
                                     content_item["text"] = ""
 
+    def _check_streaming_has_ended(self, responses_so_far: Sequence[object]) -> bool:
+        """Return True once a chat completion chunk with a non-null ``finish_reason`` has been seen.
+
+        The unified streaming hook uses this to withhold the terminal chunk in
+        ``end_of_stream_only`` mode so a block detected during final moderation
+        can replace it, instead of arriving after the client already received a
+        ``finish_reason`` and stopped reading.
+        """
+        for item in responses_so_far:
+            choices: Final = stream_item_field(item, "choices")
+            if not isinstance(choices, Sequence) or isinstance(choices, (str, bytes)):
+                continue
+            for choice in choices:
+                if stream_item_field(choice, "finish_reason") is not None:
+                    return True
+        return False
+
     def build_block_sse_chunks(
         self,
         exc: "ModifyResponseException",

diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py
--- a/litellm/llms/openai/responses/guardrail_translation/handler.py
+++ b/litellm/llms/openai/responses/guardrail_translation/handler.py
@@ -905,6 +905,7 @@
         exc: "ModifyResponseException", responses_so_far: Sequence[object]
     ) -> Sequence[ResponsesAPIStreamingResponse]:
         response_id, model, output_index = _continuation_identity(exc, responses_so_far)
+        closing_events: Final = OpenAIResponsesHandler._close_open_output_items(responses_so_far)
         item: Final = _blocked_output_item(exc)
         item_id: Final = item.id
         part: Final[_BlockedContentPart] = {"type": "output_text", "text": exc.message, "annotations": ()}
@@ -915,6 +916,7 @@
             "logprobs": None,
         }
         return (
+            *closing_events,
             OutputItemAddedEvent(
                 type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
                 output_index=output_index,
@@ -959,7 +961,110 @@
             ),
         )
 
+    @staticmethod
+    def _close_open_output_items(
+        responses_so_far: Sequence[object],
+    ) -> Sequence[ResponsesAPIStreamingResponse]:
+        """Emit close events for any output item that was ``output_item.added``
+        but never ``output_item.done`` before the block. Strict Responses SDKs
+        reject a ``response.completed`` that arrives while a prior output item
+        or content part is still open, so we close each open item (and its open
+        content parts) in ascending ``output_index`` order before appending the
+        replacement item."""
+        open_items: dict[int, dict[str, object]] = {}
+        for chunk in responses_so_far:
+            event_type = stream_item_field(chunk, "type")
+            oi = stream_item_field(chunk, "output_index")
+            if not isinstance(event_type, str) or not isinstance(oi, int):
+                continue
+            if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED.value:
+                open_items[oi] = {
+                    "item": stream_item_field(chunk, "item"),
+                    "content_parts": {},
+                }
+                continue
+            if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE.value:
+                open_items.pop(oi, None)
+                continue
+            state = open_items.get(oi)
+            if state is None:
+                continue
+            parts = cast(dict[int, dict[str, object]], state["content_parts"])
+            ci = stream_item_field(chunk, "content_index")
+            if not isinstance(ci, int):
+                continue
+            if event_type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED.value:
+                parts[ci] = {
+                    "item_id": stream_item_field(chunk, "item_id"),
+                    "text": "",
+                    "text_done": False,
+                    "part_done": False,
+                }
+            elif event_type == ResponsesAPIStreamEvents.CONTENT_PART_DONE.value and ci in parts:
+                parts[ci]["part_done"] = True
+            elif event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA.value:
+                delta = stream_item_field(chunk, "delta")
+                if isinstance(delta, str) and ci in parts:
+                    parts[ci]["text"] = cast(str, parts[ci]["text"]) + delta
+            elif event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE.value and ci in parts:
+                parts[ci]["text_done"] = True
+                text = stream_item_field(chunk, "text")
+                if isinstance(text, str):
+                    parts[ci]["text"] = text
 
+        events: list[ResponsesAPIStreamingResponse] = []
+        for oi in sorted(open_items.keys()):
+            state = open_items[oi]
+            item_obj = state["item"]
+            parts = cast(dict[int, dict[str, object]], state["content_parts"])
+            for ci in sorted(parts.keys()):
+                part_state = parts[ci]
+                item_id_raw = part_state.get("item_id")
+                item_id_str: str = item_id_raw if isinstance(item_id_raw, str) else ""
+                text_str = cast(str, part_state["text"])
+                if not part_state["text_done"]:
+                    events.append(
+                        OutputTextDoneEvent(
+                            type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
+                            item_id=item_id_str,
+                            output_index=oi,
+                            content_index=ci,
+                            text=text_str,
+                        )
+                    )
+                if not part_state["part_done"]:
+                    done_part_payload: _BlockedDoneContentPart = {
+                        "type": "output_text",
+                        "text": text_str,
+                        "annotations": (),
+                        "logprobs": None,
+                    }
+                    events.append(
+                        ContentPartDoneEvent(
+                            type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
+                            item_id=item_id_str,
+                            output_index=oi,
+                            content_index=ci,
+                            part=ContentPartDonePartOutputText.model_validate(done_part_payload),
+                        )
+                    )
+            done_item: BaseLiteLLMOpenAIResponseObject
+            if isinstance(item_obj, BaseLiteLLMOpenAIResponseObject):
+                done_item = item_obj
+            elif isinstance(item_obj, dict):
+                done_item = BaseLiteLLMOpenAIResponseObject.model_validate(item_obj)
+            else:
+                continue
+            events.append(
+                OutputItemDoneEvent(
+                    type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
+                    output_index=oi,
+                    item=done_item,
+                )
+            )
+        return tuple(events)
+
+
 class _BlockedContentPart(TypedDict):
     type: ReadOnly[str]
     text: ReadOnly[str]

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
--- 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,8 +1285,12 @@
         )
         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", (
+            "the open original output_item must be closed before the replacement is appended"
+        )
+        assert payloads[0]["output_index"] == 2
+        added_idx = types.index("response.output_item.added")
+        assert payloads[added_idx]["output_index"] == 3
         completed = payloads[-1]["response"]
         assert completed["id"] == "resp_live"
         assert completed["model"] == "gpt-5.4-mini-2026-01-01"

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
--- 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
@@ -205,6 +205,17 @@
 async def test_chat_end_of_stream_block_terminates_cleanly():
     collected = await _run_hook("/v1/chat/completions", _chat_stream(end=True), end_of_stream_only=True)
     _assert_no_error_frame(collected)
+    forwarded_finish_reasons = [
+        choice.finish_reason
+        for chunk in collected
+        if isinstance(chunk, ModelResponseStream)
+        for choice in chunk.choices
+        if choice.finish_reason is not None
+    ]
+    assert forwarded_finish_reasons == [], (
+        "the original terminal finish_reason must be withheld and replaced by the block sequence, "
+        "otherwise SDKs that stop on the first finish miss the policy text entirely"
+    )
     payloads = _sse_payloads(collected)
     assert BLOCK_MESSAGE in json.dumps(payloads)
     assert payloads[-1]["choices"][0]["finish_reason"] == "content_filter"
@@ -236,18 +247,35 @@
 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."""
+    and closes with response.completed - never a second response.created.
+
+    Also covers strict Responses SDKs that require any prior in-progress output
+    item to reach ``output_item.done`` before ``response.completed``: the open
+    item's ``output_text.done``, ``content_part.done`` and ``output_item.done``
+    events must be synthesized before the replacement item is appended."""
     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)]
     assert "response.created" in forwarded_types, "original events should have streamed before the block"
+    assert "response.output_item.done" not in forwarded_types, (
+        "the original stream stopped before closing its output item; the block sequence must close it"
+    )
     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"
+    assert block_types[:3] == [
+        "response.output_text.done",
+        "response.content_part.done",
+        "response.output_item.done",
+    ], "the block sequence must close the still-open original output item before appending the replacement"
+    for closing_payload in payloads[:3]:
+        assert closing_payload["output_index"] == 0
+    replacement_added_idx = block_types.index("response.output_item.added")
+    assert payloads[replacement_added_idx]["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

_{You can send follow-ups to the cloud agent here.}

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[veria-ai]

PR overview

All previously flagged issues have been addressed. No open security concerns remain on this pull request.

Security review

No open security issues remain on this pull request.

Fixed/addressed: 1 · PR risk: 0/10

[mateo-berri]

@greptileai

[mateo-berri]

bugbot run

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

^{Reviewed by Cursor Bugbot for commit 4567fc7. Configure here.}

[mateo-berri]

bugbot run

[cursor]

Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Usage lookup fails on typed events
  • Replaced str(...) coercion with direct equality since ResponsesAPIStreamEvents is a (str, Enum) whose members compare equal to their wire values but whose str() may return the qualified enum name.

Or push these changes by commenting:

@cursor push b4edc6ef5f

Preview (b4edc6ef5f)

diff --git a/litellm/llms/base_llm/guardrail_translation/utils.py b/litellm/llms/base_llm/guardrail_translation/utils.py
--- a/litellm/llms/base_llm/guardrail_translation/utils.py
+++ b/litellm/llms/base_llm/guardrail_translation/utils.py
@@ -171,7 +171,7 @@
         (
             response
             for item in reversed(original_response)
-            if str(stream_item_field(item, "type") or "") == "response.completed"
+            if stream_item_field(item, "type") == "response.completed"
             and (response := stream_item_field(item, "response")) is not None
         ),
         None,

_{You can send follow-ups to the cloud agent here.}

[mateo-berri]

@greptileai

[mateo-berri]

bugbot run

[cursor]

✅ Bugbot reviewed your changes and found no new issues!

Comment @cursor review or bugbot run to trigger another review on this PR

^{Reviewed by Cursor Bugbot for commit 8c7fe00. Configure here.}