From ca79f7d5cb8b08fc2768b6f8c9540c7153eb81a1 Mon Sep 17 00:00:00 2001 From: King Star Date: Wed, 9 Sep 2026 02:11:06 +0800 Subject: [PATCH] fix(streaming): fail on clean EOF without finish reason --- .../litellm_core_utils/streaming_handler.py | 23 ++++++ litellm/main.py | 2 + .../test_streaming_handler.py | 75 +++++++++++++++++++ 3 files changed, 100 insertions(+) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 6ec426d6da5..93a18189296 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -71,6 +71,7 @@ _SYNC_ITER_EXHAUSTED: Final = object() _GCHUNK_FIELDS: Final[frozenset] = frozenset(GChunk.__annotations__) _CompleteStreamingResponse = ModelResponse | TextCompletionResponse | ModelResponseStream | None _USAGE_COST_HEADER_PROVIDERS: Final[frozenset[str]] = frozenset({LlmProviders.OPENROUTER.value}) +_OPENAI_AZURE_CHAT_PROVIDERS: Final[frozenset[str]] = frozenset({LlmProviders.OPENAI.value, LlmProviders.AZURE.value}) def _next_sync_or_exhausted(it: Iterator[object]) -> object: @@ -277,11 +278,13 @@ class CustomStreamWrapper: make_call: Callable | None = None, _response_headers: dict | httpx.Headers | None = None, count_prompt_tokens: Callable[[], int] | None = None, + is_mock: bool = False, ): self.model = model self.make_call = make_call self.count_prompt_tokens = count_prompt_tokens self.custom_llm_provider = custom_llm_provider + self.is_mock = is_mock self.logging_obj: LiteLLMLoggingObject = logging_obj self.completion_stream = completion_stream self.sent_first_chunk = False @@ -1898,6 +1901,18 @@ class CustomStreamWrapper: model_response.choices[0].finish_reason = "tool_calls" return model_response + def _should_fail_on_clean_eof(self) -> bool: + return ( + not self.is_mock + and self.custom_llm_provider in _OPENAI_AZURE_CHAT_PROVIDERS + and self.received_finish_reason is None + and self.intermittent_finish_reason is None + ) + + @staticmethod + def _stream_eof_error() -> httpx.RemoteProtocolError: + return httpx.RemoteProtocolError("Stream ended before a finish_reason was received") + def _record_usage_only_chunk(self, model_response: "ModelResponseStream") -> None: """ Keep provider usage-only chunks (e.g. OpenRouter's post-finish chunk, which carries a @@ -2068,6 +2083,12 @@ class CustomStreamWrapper: return response except StopIteration: + if self._should_fail_on_clean_eof(): + error: Final = self._stream_eof_error() + eof_traceback: Final = traceback.format_exc() + self._record_partial_usage_for_failure() + threading.Thread(target=self.logging_obj.failure_handler, args=(error, eof_traceback)).start() + self._handle_stream_fallback_error(error) if self.sent_last_chunk is True: complete_streaming_response: Final = self._build_complete_streaming_response() response = self.model_response_creator() @@ -2281,6 +2302,8 @@ class CustomStreamWrapper: self.chunks.append(processed_chunk) return processed_chunk except (StopAsyncIteration, StopIteration): + if self._should_fail_on_clean_eof(): + self._log_stream_failure_and_raise(self._stream_eof_error()) return await self._finalize_completed_stream(cache_hit=cache_hit) except httpx.TimeoutException as e: # if httpx read timeout error occues traceback_exception = traceback.format_exc() diff --git a/litellm/main.py b/litellm/main.py index 63375292077..d8d1c2b79d6 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -1029,6 +1029,7 @@ def mock_completion( custom_llm_provider="openai", logging_obj=logging, count_prompt_tokens=count_prompt_tokens, + is_mock=True, ) return CustomStreamWrapper( completion_stream=mock_completion_streaming_obj( @@ -1038,6 +1039,7 @@ def mock_completion( custom_llm_provider="openai", logging_obj=logging, count_prompt_tokens=count_prompt_tokens, + is_mock=True, ) if isinstance(mock_response, litellm.MockException): raise mock_response diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 68aabc3c01e..91d953c1ac5 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -1574,6 +1574,81 @@ async def test_mid_stream_error_keeps_credentials_in_the_api_base_out(routed_api +def _incomplete_openai_chat_chunk() -> ModelResponseStream: + return ModelResponseStream( + id="chatcmpl-incomplete-eof", + created=int(time.time()), + model="gpt-5.6", + object="chat.completion.chunk", + choices=[ + StreamingChoices( + index=0, + delta=Delta(content='{"findings":[{"title":"unfinished', role="assistant"), + finish_reason=None, + ) + ], + ) + + +@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure"]) +@pytest.mark.asyncio +async def test_async_openai_chat_clean_eof_without_finish_reason_raises_midstream_fallback( + logging_obj: Logging, custom_llm_provider: str +): + """An OpenAI-compatible stream without a provider finish reason must fail at clean EOF.""" + from litellm.exceptions import MidStreamFallbackError + + async def source(): + yield _incomplete_openai_chat_chunk() + + logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider + response = CustomStreamWrapper( + completion_stream=source(), + model="gpt-5.6", + logging_obj=logging_obj, + custom_llm_provider=custom_llm_provider, + stream_options={"include_usage": True}, + ) + + first_chunk = await response.__anext__() + with pytest.raises(MidStreamFallbackError) as excinfo: + await response.__anext__() + + assert first_chunk.choices[0].delta.content == '{"findings":[{"title":"unfinished' + assert excinfo.value.generated_content == '{"findings":[{"title":"unfinished' + assert excinfo.value.is_pre_first_chunk is False + assert response.received_finish_reason is None + + +@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure"]) +def test_sync_openai_chat_clean_eof_without_finish_reason_raises_midstream_fallback( + logging_obj: Logging, custom_llm_provider: str +): + """The synchronous OpenAI-compatible stream must fail at clean EOF too.""" + from litellm.exceptions import MidStreamFallbackError + + def source(): + yield _incomplete_openai_chat_chunk() + + logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider + response = CustomStreamWrapper( + completion_stream=source(), + model="gpt-5.6", + logging_obj=logging_obj, + custom_llm_provider=custom_llm_provider, + stream_options={"include_usage": True}, + ) + + first_chunk = next(response) + with pytest.raises(MidStreamFallbackError) as excinfo: + next(response) + + assert first_chunk.choices[0].delta.content == '{"findings":[{"title":"unfinished' + assert excinfo.value.generated_content == '{"findings":[{"title":"unfinished' + assert excinfo.value.is_pre_first_chunk is False + assert response.received_finish_reason is None + + def test_streaming_handler_with_created_time_propagation( initialized_custom_stream_wrapper: CustomStreamWrapper, logging_obj: Logging ):