diff --git a/litellm/proxy/response_polling/background_streaming.py b/litellm/proxy/response_polling/background_streaming.py index bcc98175773..f1eed343892 100644 --- a/litellm/proxy/response_polling/background_streaming.py +++ b/litellm/proxy/response_polling/background_streaming.py @@ -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, diff --git a/tests/proxy_unit_tests/test_response_polling_handler.py b/tests/proxy_unit_tests/test_response_polling_handler.py index c5f3d7c6f45..54dbe1f024e 100644 --- a/tests/proxy_unit_tests/test_response_polling_handler.py +++ b/tests/proxy_unit_tests/test_response_polling_handler.py @@ -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"""