fix(streaming): fail on clean EOF without finish reason

This commit is contained in:
King Star 2026-09-09 02:11:06 +08:00 • committed by yucheng-berri
parent f0ae445cc4
commit ca79f7d5cb
3 changed files with 100 additions and 0 deletions

View file

@ -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()

View file

@ -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

View file

@ -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
):