scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Fix background streaming silently dropping output and errors

Browse source 
The background streaming task had several bugs that caused it to report
status "completed" with empty output when using non-native providers
(Anthropic, Bedrock) via the LiteLLMCompletionStreamingIterator:

1. Text deltas silently dropped: The adapter emits output_item.added with
   content: [] but skips content_part.added. The text delta handler required
   content parts to already exist, so deltas were discarded.

2. content_part.done same empty-list issue: Same bounds check failure when
   the content list was empty.

3. SSE error chunks silently ignored: async_data_generator emits
   {"error": {...}} with no "type" field on exceptions. The event dispatcher
   matched nothing, swallowing the error entirely.

4. Missing terminal event falsely reported "completed": When the stream
   ended without response.completed/failed/incomplete (e.g. due to a
   swallowed error), the fallback was "completed" instead of "failed".

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Committed-By-Agent: claude
This commit is contained in:
Xianzong Xie 2026-03-24 13:51:45 -07:00
parent 25feae9f0f
commit ffa727233a
2 changed files with 165 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

68
litellm/proxy/response_polling/background_streaming.py
View file

@ -166,6 +166,23 @@ async def background_streaming_task( # noqa: PLR0915
try:
event = json.loads(chunk_data)
# Handle SSE error chunks from async_data_generator.
# These are {"error": {...}} dicts with no "type" field,
# emitted when the streaming iterator raises an exception.
if "error" in event and "type" not in event:
error_info = event["error"]
terminal_status = "failed"
terminal_error = {
"type": error_info.get("type", "internal_error"),
"message": error_info.get("message", str(error_info)),
"code": error_info.get("code", "streaming_error"),
}
verbose_proxy_logger.error(
f"Received SSE error for {polling_id}: {terminal_error['message']}"
)
continue
event_type = event.get("type", "")
# Process different event types based on OpenAI streaming spec
@ -204,16 +221,17 @@ async def background_streaming_task( # noqa: PLR0915
accumulated_text[key] += delta
# Update the content in output_items
if "content" in output_items[item_id]:
content_list = output_items[item_id]["content"]
if content_index < len(content_list):
# Update existing content part with accumulated text
if isinstance(
content_list[content_index], dict
):
content_list[content_index][
"text"
] = accumulated_text[key]
if "content" not in output_items[item_id]:
output_items[item_id]["content"] = []
content_list = output_items[item_id]["content"]
# Auto-create content part if missing.
# Some providers (e.g. Anthropic via the litellm
# completion adapter) don't emit
# response.content_part.added before text deltas.
while len(content_list) <= content_index:
content_list.append({"type": "output_text", "text": ""})
if isinstance(content_list[content_index], dict):
content_list[content_index]["text"] = accumulated_text[key]
state_dirty = True
elif event_type == "response.content_part.done":
@ -223,11 +241,13 @@ async def background_streaming_task( # noqa: PLR0915
content_index = event.get("content_index", 0)
if item_id and item_id in output_items:
# Update with final content from event
if "content" in output_items[item_id]:
content_list = output_items[item_id]["content"]
if content_index < len(content_list):
content_list[content_index] = content_part
if "content" not in output_items[item_id]:
output_items[item_id]["content"] = []
content_list = output_items[item_id]["content"]
# Grow content list if needed (mirrors text delta fix)
while len(content_list) <= content_index:
content_list.append({"type": "output_text", "text": ""})
content_list[content_index] = content_part
state_dirty = True
elif event_type == "response.output_item.done":
@ -318,9 +338,21 @@ async def background_streaming_task( # noqa: PLR0915
# Final flush to ensure all accumulated state is saved
await flush_state_if_needed(force=True)
# Use the terminal status from the stream, default to "completed"
final_status = terminal_status or "completed"
# Use the terminal status from the stream. If no terminal event was
# received (e.g. the stream errored before response.completed), treat
# as failed rather than silently reporting success.
if terminal_status is None and terminal_error is None:
verbose_proxy_logger.warning(
f"No terminal event received for {polling_id}; "
"marking as failed (stream may have ended prematurely)"
)
terminal_status = "failed"
terminal_error = {
"type": "internal_error",
"message": "Stream ended without a terminal event (response.completed/failed/incomplete)",
"code": "missing_terminal_event",
}
final_status = terminal_status or "failed"
await polling_handler.update_state(
polling_id=polling_id,

117
tests/proxy_unit_tests/test_response_polling_handler.py
View file

@ -1560,8 +1560,9 @@ class TestBackgroundStreamingTerminalEvents:
assert final_call.kwargs["status"] == "incomplete"
@pytest.mark.asyncio
async def test_no_terminal_event_defaults_to_completed(self):
"""Test that when no terminal event is received, status defaults to completed"""
async def test_no_terminal_event_defaults_to_failed(self):
"""Test that when no terminal event is received, status defaults to failed
with a descriptive error (stream ended prematurely)."""
from litellm.proxy.response_polling.background_streaming import (
background_streaming_task,
)
@ -1574,6 +1575,104 @@ class TestBackgroundStreamingTerminalEvents:
handler = AsyncMock(spec=ResponsePollingHandler)
kwargs = _make_background_streaming_kwargs("poll_6", handler)
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
assert final_call.kwargs["status"] == "failed"
assert final_call.kwargs["error"]["code"] == "missing_terminal_event"
assert final_call.kwargs["error"]["type"] == "internal_error"
@pytest.mark.asyncio
async def test_sse_error_chunk_sets_failed_status(self):
"""Test that an SSE error chunk (no 'type' field) from async_data_generator
results in failed status with the error extracted."""
from litellm.proxy.response_polling.background_streaming import (
background_streaming_task,
)
events = [
{"type": "response.in_progress"},
# Error chunk emitted by async_data_generator — no "type" field
{
"error": {
"message": "AnthropicException: overloaded",
"type": "internal_error",
"code": 529,
}
},
]
mock_response = _make_sse_stream(events)
handler = AsyncMock(spec=ResponsePollingHandler)
kwargs = _make_background_streaming_kwargs("poll_7", handler)
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
assert final_call.kwargs["status"] == "failed"
assert final_call.kwargs["error"]["message"] == "AnthropicException: overloaded"
assert final_call.kwargs["error"]["code"] == 529
@pytest.mark.asyncio
async def test_text_delta_auto_creates_content_parts(self):
"""Test that text deltas auto-create content parts when no
ContentPartAddedEvent was emitted (Anthropic/Bedrock adapter path)."""
from litellm.proxy.response_polling.background_streaming import (
background_streaming_task,
)
events = [
{"type": "response.in_progress"},
# OutputItemAdded with empty content (like LiteLLMCompletionStreamingIterator)
{
"type": "response.output_item.added",
"item": {"id": "item_1", "type": "message", "content": []},
},
# Text deltas without prior ContentPartAdded
{
"type": "response.output_text.delta",
"item_id": "item_1",
"content_index": 0,
"delta": "Hello",
},
{
"type": "response.output_text.delta",
"item_id": "item_1",
"content_index": 0,
"delta": " world",
},
{
"type": "response.completed",
"response": {
"id": "resp_123",
"status": "completed",
"model": "claude-sonnet-4-20250514",
"output": [
{
"id": "item_1",
"type": "message",
"content": [{"type": "output_text", "text": "Hello world"}],
}
],
"usage": {"input_tokens": 5, "output_tokens": 2},
},
},
]
mock_response = _make_sse_stream(events)
handler = AsyncMock(spec=ResponsePollingHandler)
kwargs = _make_background_streaming_kwargs("poll_8", handler)
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
@ -1585,6 +1684,20 @@ class TestBackgroundStreamingTerminalEvents:
final_call = handler.update_state.call_args_list[-1]
assert final_call.kwargs["status"] == "completed"
# Verify intermediate flush captured the auto-created content.
# Find a flush call that included output with populated text.
output_calls = [
c for c in handler.update_state.call_args_list
if "output" in c.kwargs and c.kwargs["output"]
]
# At minimum, the final response.completed should have updated output
assert len(output_calls) > 0
# The final output should have content with "Hello world"
last_output = output_calls[-1].kwargs["output"]
item_1 = [i for i in last_output if i.get("id") == "item_1"][0]
assert len(item_1["content"]) == 1
assert item_1["content"][0]["text"] == "Hello world"
class TestEdgeCases:
"""Test edge cases and error scenarios"""