From ca79f7d5cb8b08fc2768b6f8c9540c7153eb81a1 Mon Sep 17 00:00:00 2001 From: King Star Date: Wed, 9 Sep 2026 02:11:06 +0800 Subject: [PATCH 01/22] 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 ): From 72488ab579ec9048585cd2a9e146a17c1e744856 Mon Sep 17 00:00:00 2001 From: King Star Date: Wed, 9 Sep 2026 02:44:00 +0800 Subject: [PATCH 02/22] fix(streaming): cover shared OpenAI-compatible EOF paths --- litellm/litellm_core_utils/streaming_handler.py | 4 +++- tests/unit/litellm_core_utils/test_streaming_handler.py | 8 ++++---- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 93a18189296..2b6e827eed6 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -285,6 +285,7 @@ class CustomStreamWrapper: self.count_prompt_tokens = count_prompt_tokens self.custom_llm_provider = custom_llm_provider self.is_mock = is_mock + self.uses_openai_chat_stream = False self.logging_obj: LiteLLMLoggingObject = logging_obj self.completion_stream = completion_stream self.sent_first_chunk = False @@ -1518,6 +1519,7 @@ class CustomStreamWrapper: if response_obj["is_finished"]: self.received_finish_reason = response_obj["finish_reason"] else: # openai / azure chat model + self.uses_openai_chat_stream = True if isinstance(chunk, ChatCompletionChunk): self._openai_handler_receives_chat_chunks = True if self.custom_llm_provider in [ @@ -1904,7 +1906,7 @@ class CustomStreamWrapper: 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.custom_llm_provider in _OPENAI_AZURE_CHAT_PROVIDERS or self.uses_openai_chat_stream) and self.received_finish_reason is None and self.intermittent_finish_reason is None ) diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 91d953c1ac5..eec2a94bc78 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -1590,9 +1590,9 @@ def _incomplete_openai_chat_chunk() -> ModelResponseStream: ) -@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure"]) +@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure", "hosted_vllm"]) @pytest.mark.asyncio -async def test_async_openai_chat_clean_eof_without_finish_reason_raises_midstream_fallback( +async def test_async_openai_compatible_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.""" @@ -1620,8 +1620,8 @@ async def test_async_openai_chat_clean_eof_without_finish_reason_raises_midstrea 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( +@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure", "hosted_vllm"]) +def test_sync_openai_compatible_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 bdb52437e6d5fd6f1af35789ac08d33d37dcd63a Mon Sep 17 00:00:00 2001 From: King Star Date: Wed, 9 Sep 2026 03:04:47 +0800 Subject: [PATCH 03/22] test(router): complete mocked streaming responses --- tests/unit/test_router/test_router.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index d645ab4566d..d5e3d73f2a2 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -21505,6 +21505,9 @@ _SSE_CHUNKS: Final[tuple[bytes, ...]] = tuple( b'data: {"id":"c","object":"chat.completion.chunk","created":1,"model":"gpt-5.6",' b'"choices":[{"index":0,"delta":{"content":"x"},"finish_reason":null}]}\n\n' for _ in range(5) +) + ( + b'data: {"id":"c","object":"chat.completion.chunk","created":1,"model":"gpt-5.6",' + b'"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n', ) From 127e0475fee00793d52d76b76a24625e42d42410 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:44:14 -0700 Subject: [PATCH 04/22] fix(streaming): keep the EOF error message and cover Anthropic clean EOF The request-less RemoteProtocolError tripped exception mapping into 'The .request property has not been set.' for non-OpenAI providers. Adds Anthropic and generic-chunk unit tests, an integration test across chat, the /v1/messages bridge and the /v1/responses bridge, and moves main's token-estimate test onto a finished stream with a sibling that checks the cut stream still bills the estimate on its failure log. --- .../litellm_core_utils/streaming_handler.py | 7 +- ..._stream_clean_eof_without_finish_reason.py | 200 ++++++++++++++++++ .../test_streaming_handler.py | 181 ++++++++++++++-- 3 files changed, 374 insertions(+), 14 deletions(-) create mode 100644 tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 2b6e827eed6..0911c4bbdf4 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -1913,7 +1913,10 @@ class CustomStreamWrapper: @staticmethod def _stream_eof_error() -> httpx.RemoteProtocolError: - return httpx.RemoteProtocolError("Stream ended before a finish_reason was received") + return httpx.RemoteProtocolError( + "Stream ended before a finish_reason was received", + request=httpx.Request(method="POST", url="https://docs.litellm.ai/docs"), + ) def _record_usage_only_chunk(self, model_response: "ModelResponseStream") -> None: """ @@ -2090,6 +2093,7 @@ class CustomStreamWrapper: 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.close() self._handle_stream_fallback_error(error) if self.sent_last_chunk is True: complete_streaming_response: Final = self._build_complete_streaming_response() @@ -2305,6 +2309,7 @@ class CustomStreamWrapper: return processed_chunk except (StopAsyncIteration, StopIteration): if self._should_fail_on_clean_eof(): + await self.aclose() 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 diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py new file mode 100644 index 00000000000..f1dad7e8844 --- /dev/null +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -0,0 +1,200 @@ +import json +import uuid +from collections.abc import Callable, Mapping +from typing import Final, Literal + +import pytest +from integration._support.anthropic_sse import CONTENT_BLOCK_START, CONTENT_TAIL, message_start, text_delta +from integration._support.client import Gateway, string_value +from integration._support.openai_wire import answering_model_discovery +from integration._support.responses_stream import data_frames, spend_row, sse_events +from integration._support.wire import Reply, Request, wire_server +from pydantic import JsonValue + +Ending = Literal["ending-complete", "ending-clean-eof", "ending-done-without-finish"] +Family = Literal["openai", "anthropic", "hosted_vllm"] + +_ENDINGS: Final[tuple[Ending, ...]] = ("ending-complete", "ending-clean-eof", "ending-done-without-finish") +_PARTIAL: Final = ('{"findings":[{"title":', '"unfinished') +_BACKEND: Final = "gpt-4o-mini" +_CLAUDE: Final = "claude-sonnet-4-5" +_PROVIDER_KEY: Final = "synthetic-provider-key" +_CALL_ID_HEADER: Final = "x-litellm-call-id" +_USAGE: Final = {"prompt_tokens": 9, "completion_tokens": 7, "total_tokens": 16} +_TARGETS: Final[Mapping[Family, str]] = { + "openai": "/chat/completions", + "hosted_vllm": "/chat/completions", + "anthropic": "/v1/messages", +} +_DEPLOYMENTS: Final[Mapping[Family, str]] = { + "openai": f"openai/{_BACKEND}", + "hosted_vllm": f"hosted_vllm/{_BACKEND}", + "anthropic": f"anthropic/{_CLAUDE}", +} + + +def _data(frame: Mapping[str, JsonValue]) -> bytes: + return b"data: " + json.dumps(frame).encode() + b"\n\n" + + +def _chat_frames(identity: str, ending: Ending) -> tuple[bytes, ...]: + chunk: Final = {"id": identity, "object": "chat.completion.chunk", "created": 1, "model": _BACKEND} + content: Final = ( + _data({**chunk, "choices": [{"index": 0, "delta": {"role": "assistant", "content": _PARTIAL[0]}}]}), + _data({**chunk, "choices": [{"index": 0, "delta": {"content": _PARTIAL[1]}, "finish_reason": None}]}), + ) + match ending: + case "ending-complete": + finish: Final = _data({**chunk, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}) + usage: Final = _data({**chunk, "choices": [], "usage": _USAGE}) + return (*content, finish + usage + b"data: [DONE]\n\n") + case "ending-clean-eof": + return content + case "ending-done-without-finish": + return (*content, b"data: [DONE]\n\n") + + +def _anthropic_frames(identity: str, ending: Ending) -> tuple[bytes, ...]: + content: Final = (message_start(identity, _CLAUDE), CONTENT_BLOCK_START, *map(text_delta, _PARTIAL)) + return (*content, CONTENT_TAIL) if ending == "ending-complete" else content + + +def _ending(request: Request) -> Ending: + return next(ending for ending in _ENDINGS if ending.encode() in request.body) + + +def _upstream(family: Family) -> Callable[[Request], Reply]: + def respond(request: Request) -> Reply: + assert request.method == "POST" and request.target.endswith(_TARGETS[family]), request.target + identity: Final = f"upstream-{uuid.uuid4().hex}" + frames: Final = ( + _anthropic_frames(identity, _ending(request)) + if family == "anthropic" + else _chat_frames(identity, _ending(request)) + ) + return Reply(content_type="text/event-stream", chunks=frames) + + return answering_model_discovery(respond) + + +def _prompt(ending: Ending) -> str: + return f"Return JSON {ending} {uuid.uuid4().hex}" + + +def _post(gateway: Gateway, path: str, body: Mapping[str, JsonValue]) -> tuple[str, str]: + response: Final = gateway.request("POST", path, {**body, "stream": True, "num_retries": 0}) + assert response.status_code == 200, response.text + return response.text, response.headers[_CALL_ID_HEADER] + + +def _chat(gateway: Gateway, model: str, ending: Ending) -> tuple[str, str]: + return _post( + gateway, + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": _prompt(ending)}], + "stream_options": {"include_usage": True}, + }, + ) + + +def _finish_reasons(text: str) -> tuple[str, ...]: + return tuple( + string_value(choice["finish_reason"]) + for frame in data_frames(text) + for choice in frame.get("choices") or [] + if isinstance(choice, dict) and choice.get("finish_reason") + ) + + +def _chat_text(text: str) -> str: + return "".join( + string_value(delta.get("content") or "") + for frame in data_frames(text) + for choice in frame.get("choices") or [] + if isinstance(choice, dict) + for delta in (choice.get("delta") or {},) + if isinstance(delta, dict) + ) + + +def _chat_errors(text: str) -> tuple[Mapping[str, JsonValue], ...]: + return tuple(frame for frame in data_frames(text) if "error" in frame) + + +def _event_types(text: str) -> tuple[str, ...]: + return tuple(event for event, _ in sse_events(text)) + + +@pytest.mark.parametrize("family", ["openai", "anthropic"]) +def test_chat_stream_that_ends_cleanly_without_a_finish_reason_reaches_the_client_as_an_error( + gateway: Gateway, family: Family +) -> None: + with wire_server(_upstream(family)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_DEPLOYMENTS[family], api_base=wire.url, api_key=_PROVIDER_KEY) + text, call_id = _chat(gateway, model, "ending-clean-eof") + assert _chat_text(text) == "".join(_PARTIAL), text + assert _finish_reasons(text) == (), text + assert len(_chat_errors(text)) == 1, text + assert spend_row(call_id)["status"] == "failure" + + +def test_chat_stream_that_sends_done_without_a_finish_reason_reaches_the_client_as_an_error(gateway: Gateway) -> None: + with wire_server(_upstream("openai")) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_DEPLOYMENTS["openai"], api_base=wire.url, api_key=_PROVIDER_KEY) + text, call_id = _chat(gateway, model, "ending-done-without-finish") + assert _chat_text(text) == "".join(_PARTIAL), text + assert _finish_reasons(text) == (), text + assert len(_chat_errors(text)) == 1, text + assert spend_row(call_id)["status"] == "failure" + + +@pytest.mark.parametrize("family", ["openai", "anthropic"]) +def test_chat_stream_with_a_provider_finish_reason_still_completes(gateway: Gateway, family: Family) -> None: + with wire_server(_upstream(family)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_DEPLOYMENTS[family], api_base=wire.url, api_key=_PROVIDER_KEY) + text, call_id = _chat(gateway, model, "ending-complete") + assert _chat_text(text) == "".join(_PARTIAL), text + assert _finish_reasons(text) == ("stop",), text + assert _chat_errors(text) == (), text + assert text.rstrip().endswith("data: [DONE]"), text + assert spend_row(call_id)["status"] == "success" + + +@pytest.mark.parametrize(("ending", "terminal"), [("ending-complete", True), ("ending-clean-eof", False)]) +def test_messages_over_a_chat_bridged_deployment_reports_end_turn_only_for_a_finished_stream( + gateway: Gateway, ending: Ending, terminal: bool +) -> None: + with wire_server(_upstream("hosted_vllm")) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_DEPLOYMENTS["hosted_vllm"], api_base=wire.url, api_key=_PROVIDER_KEY) + text, call_id = _post( + gateway, + "/v1/messages", + {"model": model, "max_tokens": 64, "messages": [{"role": "user", "content": _prompt(ending)}]}, + ) + types: Final = _event_types(text) + stop_reasons: Final = tuple( + data["delta"]["stop_reason"] + for event, data in sse_events(text) + if event == "message_delta" and isinstance(data.get("delta"), dict) + ) + assert types[0] == "message_start", types + assert ("message_stop" in types) is terminal, types + assert ("error" in types) is not terminal, types + assert stop_reasons == (("end_turn",) if terminal else ()), types + assert spend_row(call_id)["status"] == ("success" if terminal else "failure") + + +@pytest.mark.parametrize(("ending", "terminal"), [("ending-complete", True), ("ending-clean-eof", False)]) +def test_responses_over_a_chat_bridged_anthropic_deployment_completes_only_a_finished_stream( + gateway: Gateway, ending: Ending, terminal: bool +) -> None: + with wire_server(_upstream("anthropic")) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_DEPLOYMENTS["anthropic"], api_base=wire.url, api_key=_PROVIDER_KEY) + text, call_id = _post(gateway, "/v1/responses", {"model": model, "input": _prompt(ending)}) + types: Final = tuple(string_value(frame.get("type") or "") for frame in data_frames(text)) + assert types[:1] == ("response.created",), text + assert ("response.completed" in types) is terminal, text + assert (types[-1] in {"error", "response.failed"}) is not terminal, text + assert spend_row(call_id)["status"] == ("success" if terminal else "failure") diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index eec2a94bc78..7aa0f7b01ad 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -27,6 +27,7 @@ from litellm.litellm_core_utils.streaming_handler import ( _ProviderChunkEarlyReturn, _ProviderChunkParsed, ) +from litellm.llms.anthropic.chat.handler import ModelResponseIterator as AnthropicModelResponseIterator from litellm.llms.custom_httpx.http_handler import HTTPHandler from litellm.responses.mcp.chat_completions_handler import _add_mcp_metadata_to_response from litellm.types.utils import ( @@ -1596,7 +1597,6 @@ async def test_async_openai_compatible_chat_clean_eof_without_finish_reason_rais 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() @@ -1625,7 +1625,6 @@ def test_sync_openai_compatible_chat_clean_eof_without_finish_reason_raises_mids 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() @@ -1649,6 +1648,117 @@ def test_sync_openai_compatible_chat_clean_eof_without_finish_reason_raises_mids assert response.received_finish_reason is None +_ANTHROPIC_PARTIAL_EVENTS: Final[tuple[dict[str, object], ...]] = ( + { + "type": "message_start", + "message": { + "id": "msg_incomplete_eof", + "type": "message", + "role": "assistant", + "model": "claude-sonnet-4-5", + "content": [], + "stop_reason": None, + "usage": {"input_tokens": 9, "output_tokens": 1}, + }, + }, + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": '{"findings":[{"title":"unfinished'}, + }, +) +_ANTHROPIC_TERMINAL_EVENTS: Final[tuple[dict[str, object], ...]] = ( + {"type": "content_block_stop", "index": 0}, + {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 7}}, + {"type": "message_stop"}, +) + + +def _anthropic_sse_lines(events: Sequence[Mapping[str, object]]) -> list[str]: + return [line for event in events for line in (f"event: {event['type']}", f"data: {json.dumps(event)}", "")] + + +def _anthropic_stream( + events: Sequence[Mapping[str, object]], sync_mode: bool, logging_obj: Logging +) -> CustomStreamWrapper: + lines: Final = _anthropic_sse_lines(events) + + async def async_lines() -> AsyncIterator[str]: + for line in lines: + yield line + + return CustomStreamWrapper( + completion_stream=AnthropicModelResponseIterator( + streaming_response=iter(lines) if sync_mode else async_lines(), sync_stream=sync_mode + ), + model="claude-sonnet-4-5", + custom_llm_provider="anthropic", + logging_obj=logging_obj, + ) + + +async def _collect(wrapper: CustomStreamWrapper, sync_mode: bool) -> list[ModelResponseStream]: + if sync_mode: + return list(wrapper) + return [chunk async for chunk in wrapper] + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_anthropic_stream_that_ends_before_message_delta_fails_with_the_eof_message( + sync_mode: bool, logging_obj: Logging +): + wrapper: Final = _anthropic_stream(_ANTHROPIC_PARTIAL_EVENTS, sync_mode, logging_obj) + + with pytest.raises(MidStreamFallbackError) as excinfo: + await _collect(wrapper, sync_mode) + + assert excinfo.value.generated_content == '{"findings":[{"title":"unfinished' + assert excinfo.value.is_pre_first_chunk is False + assert isinstance(excinfo.value.original_exception, litellm.APIConnectionError) + assert "AnthropicException - Stream ended before a finish_reason was received" in str(excinfo.value) + assert wrapper.received_finish_reason is None + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_anthropic_stream_with_message_stop_ends_with_its_finish_reason(sync_mode: bool, logging_obj: Logging): + wrapper: Final = _anthropic_stream( + (*_ANTHROPIC_PARTIAL_EVENTS, *_ANTHROPIC_TERMINAL_EVENTS), sync_mode, logging_obj + ) + + received: Final = await _collect(wrapper, sync_mode) + + assert [c.choices[0].finish_reason for c in received if c.choices and c.choices[0].finish_reason] == ["stop"] + assert "".join(c.choices[0].delta.content or "" for c in received if c.choices) == ( + '{"findings":[{"title":"unfinished' + ) + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_generic_chunk_stream_without_a_finish_reason_keeps_the_default_stop( + sync_mode: bool, logging_obj: Logging +): + """Providers that stream GenericStreamingChunks and never mark the last one finished still end with stop.""" + chunks: Final = [ + {"text": text, "tool_use": None, "is_finished": False, "finish_reason": "", "usage": None, "index": 0} + for text in ("partial", " reply") + ] + wrapper: Final = CustomStreamWrapper( + completion_stream=ModelResponseListIterator(model_responses=chunks), + model="sagemaker/tgi-endpoint", + custom_llm_provider="sagemaker", + logging_obj=logging_obj, + ) + + received: Final = await _collect(wrapper, sync_mode) + + assert [c.choices[0].finish_reason for c in received if c.choices and c.choices[0].finish_reason] == ["stop"] + assert "".join(c.choices[0].delta.content or "" for c in received if c.choices) == "partial reply" + + def test_streaming_handler_with_created_time_propagation( initialized_custom_stream_wrapper: CustomStreamWrapper, logging_obj: Logging ): @@ -2899,11 +3009,8 @@ def test_usage_chunk_after_finish_reason_updates_hidden_params(logging_obj): @pytest.mark.parametrize("sync_mode", [True, False]) -@pytest.mark.parametrize("ends_with_finish_chunk", [True, False]) @pytest.mark.asyncio -async def test_stream_without_provider_usage_falls_back_to_the_token_estimate( - sync_mode: bool, ends_with_finish_chunk: bool -) -> None: +async def test_stream_without_provider_usage_falls_back_to_the_token_estimate(sync_mode: bool) -> None: model: Final = "gpt-4o" messages: Final = [{"role": "user", "content": "Write a long sentence about a fox. " * 30}] text: Final = "The quick brown fox jumps over the lazy dog. " * 20 @@ -2920,9 +3027,7 @@ async def test_stream_without_provider_usage_falls_back_to_the_token_estimate( choices=[StreamingChoices(index=0, delta=Delta(content=""), finish_reason="stop")], ) wrapper: Final = CustomStreamWrapper( - completion_stream=ModelResponseListIterator( - model_responses=[content_chunk, finish_chunk] if ends_with_finish_chunk else [content_chunk] - ), + completion_stream=ModelResponseListIterator(model_responses=[content_chunk, finish_chunk]), model=model, custom_llm_provider="openai", logging_obj=Logging( @@ -2961,6 +3066,55 @@ async def test_stream_without_provider_usage_falls_back_to_the_token_estimate( assert litellm.completion_cost(completion_response=assembled) == pytest.approx(spend) +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_stream_cut_before_a_finish_chunk_bills_the_token_estimate_on_its_failure(sync_mode: bool) -> None: + model: Final = "gpt-4o" + messages: Final = [{"role": "user", "content": "Write a long sentence about a fox. " * 30}] + text: Final = "The quick brown fox jumps over the lazy dog. " * 20 + logging_obj: Final = Logging( + model=model, + messages=messages, + stream=True, + call_type="completion", + start_time=datetime.datetime(2025, 3, 3, 21, 38, 10), + litellm_call_id="cut-no-usage-call", + function_id="cut-no-usage-fn", + ) + logging_obj.update_environment_variables( + model=model, user=None, optional_params={}, litellm_params={"litellm_call_id": "cut-no-usage-call"} + ) + wrapper: Final = CustomStreamWrapper( + completion_stream=ModelResponseListIterator( + model_responses=[ + ModelResponseStream( + id="chatcmpl-cut-no-usage", + created=1741037890, + model=model, + choices=[StreamingChoices(index=0, delta=Delta(role="assistant", content=text))], + ) + ] + ), + model=model, + custom_llm_provider="openai", + logging_obj=logging_obj, + stream_options=None, + ) + + with pytest.raises(MidStreamFallbackError): + await _collect(wrapper, sync_mode) + + expected: Final = ( + litellm.token_counter(model=model, messages=messages), + litellm.token_counter(model=model, text=text, count_response_tokens=True), + ) + usage: Final = logging_obj.model_call_details["combined_usage_object"] + assert (usage.prompt_tokens, usage.completion_tokens) == expected + rates: Final = litellm.model_cost[model] + spend: Final = expected[0] * rates["input_cost_per_token"] + expected[1] * rates["output_cost_per_token"] + assert logging_obj.model_call_details["response_cost"] == pytest.approx(spend) + + @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio async def test_stream_with_usage_after_the_finish_chunk_bills_only_the_reported_usage(sync_mode: bool) -> None: @@ -7461,10 +7615,11 @@ async def test_no_synthetic_finish_reason_logged_when_provider_sent_none(sync_mo custom_llm_provider="bedrock", logging_obj=logging_obj, ) - if sync_mode: - list(response) - else: - [c async for c in response] + with pytest.raises(MidStreamFallbackError): + if sync_mode: + list(response) + else: + [c async for c in response] assert response.received_finish_reason is None assert all(not (c.choices and c.choices[0].finish_reason) for c in response.chunks) From 3f924b781189ed79cd7ba479a7afc9cc03bc0b09 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:40:55 -0700 Subject: [PATCH 05/22] fix(streaming): count a finish reason any chunk sent, keep LangGraph async streams whole Gemini marks every chunk unfinished and a later usage chunk reset intermittent_finish_reason, so a finished Gemini stream failed at EOF. LangGraph's __aiter__ opened a new aiter_lines on every wrapper call, which hit StreamConsumed after the first chunk and now surfaced as a false EOF error instead of a silent truncation. --- .../litellm_core_utils/streaming_handler.py | 3 ++ litellm/llms/langgraph/chat/sse_iterator.py | 5 ++- .../test_streaming_handler.py | 45 +++++++++++++++++++ .../chat/test_langgraph_transformation.py | 25 +++++++++++ 4 files changed, 76 insertions(+), 2 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 0911c4bbdf4..70bad6395ef 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -307,6 +307,7 @@ class CustomStreamWrapper: self._provider_response_model: str | None = None self.received_finish_reason: str | None = None self.intermittent_finish_reason: str | None = None # finish reasons that show up mid-stream + self.received_any_finish_reason = False # a later chunk can reset intermittent_finish_reason self.special_tokens = [ "<|assistant|>", "<|system|>", @@ -1534,6 +1535,7 @@ class CustomStreamWrapper: return _ProviderChunkEarlyReturn(None) completion_obj["content"] = response_obj["text"] self.intermittent_finish_reason = response_obj.get("finish_reason", None) + self.received_any_finish_reason |= self.intermittent_finish_reason is not None if response_obj["is_finished"]: if response_obj["finish_reason"] == "error": raise Exception( @@ -1909,6 +1911,7 @@ class CustomStreamWrapper: and (self.custom_llm_provider in _OPENAI_AZURE_CHAT_PROVIDERS or self.uses_openai_chat_stream) and self.received_finish_reason is None and self.intermittent_finish_reason is None + and not self.received_any_finish_reason ) @staticmethod diff --git a/litellm/llms/langgraph/chat/sse_iterator.py b/litellm/llms/langgraph/chat/sse_iterator.py index 499cbab3b7b..6b2f0c2a3a8 100644 --- a/litellm/llms/langgraph/chat/sse_iterator.py +++ b/litellm/llms/langgraph/chat/sse_iterator.py @@ -37,8 +37,9 @@ class LangGraphSSEStreamIterator: return self def __aiter__(self): - """Initialize async iteration.""" - self.async_line_iterator = self.response.aiter_lines() + """Initialize async iteration once; CustomStreamWrapper re-enters `async for` on every chunk.""" + if self.async_line_iterator is None: + self.async_line_iterator = self.response.aiter_lines() return self def _parse_sse_line(self, line: str) -> ModelResponseStream | None: diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 7aa0f7b01ad..2d1c2a581cf 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -29,6 +29,9 @@ from litellm.litellm_core_utils.streaming_handler import ( ) from litellm.llms.anthropic.chat.handler import ModelResponseIterator as AnthropicModelResponseIterator from litellm.llms.custom_httpx.http_handler import HTTPHandler +from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import ( + ModelResponseIterator as GeminiModelResponseIterator, +) from litellm.responses.mcp.chat_completions_handler import _add_mcp_metadata_to_response from litellm.types.utils import ( CompletionTokensDetailsWrapper, @@ -1759,6 +1762,48 @@ async def test_generic_chunk_stream_without_a_finish_reason_keeps_the_default_st assert "".join(c.choices[0].delta.content or "" for c in received if c.choices) == "partial reply" +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_gemini_stream_with_a_usage_chunk_after_its_finish_reason_still_completes( + sync_mode: bool, logging_obj: Logging +): + """Gemini marks every chunk unfinished, so its finishReason only shows up mid-stream before the usage chunk.""" + lines: Final = [ + f"data: {json.dumps(event)}" + for event in ( + {"candidates": [{"content": {"parts": [{"text": "Hello"}], "role": "model"}, "index": 0}]}, + { + "candidates": [ + {"content": {"parts": [{"text": " there"}], "role": "model"}, "finishReason": "STOP", "index": 0} + ] + }, + {"usageMetadata": {"promptTokenCount": 3, "candidatesTokenCount": 2, "totalTokenCount": 5}}, + ) + ] + + async def async_lines() -> AsyncIterator[str]: + for line in lines: + yield line + + logging_obj.update_environment_variables( + model="gemini-2.5-flash", user=None, optional_params={}, litellm_params={"litellm_call_id": "gemini-tail"} + ) + iterator: Final = GeminiModelResponseIterator( + streaming_response=iter(lines) if sync_mode else async_lines(), sync_stream=sync_mode, logging_obj=logging_obj + ) + wrapper: Final = CustomStreamWrapper( + completion_stream=iter(iterator) if sync_mode else iterator, + model="gemini-2.5-flash", + custom_llm_provider="gemini", + logging_obj=logging_obj, + ) + + received: Final = await _collect(wrapper, sync_mode) + + assert "".join(c.choices[0].delta.content or "" for c in received if c.choices) == "Hello there" + assert [c.choices[0].finish_reason for c in received if c.choices and c.choices[0].finish_reason] == ["stop"] + + def test_streaming_handler_with_created_time_propagation( initialized_custom_stream_wrapper: CustomStreamWrapper, logging_obj: Logging ): diff --git a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py index 238f7dd2f26..6612c71f436 100644 --- a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py +++ b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py @@ -268,3 +268,28 @@ async def test_langgraph_streaming_acompletion_without_a_client_posts_through_th "assistant_id": "agent", "input": {"messages": [{"role": "human", "content": "What is 25 * 4?"}]}, } + + +@pytest.mark.asyncio +async def test_langgraph_async_stream_delivers_every_message_event_before_its_stop(monkeypatch): + async def events(): + yield b'data: ["messages", [[{"type": "ai", "content": "25 * 4"}, {}]]]\n\n' + yield b'data: ["messages", [[{"type": "ai", "content": " is 100"}, {}]]]\n\n' + yield b'data: ["metadata", {"run_id": "run-1"}]\n\n' + + cached_client = get_async_httpx_client(llm_provider="langgraph", params={}) + monkeypatch.setattr( + cached_client, + "client", + httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200, content=events()))), + ) + + response = await litellm.acompletion( + model="langgraph/agent", + messages=[{"role": "user", "content": "What is 25 * 4?"}], + api_base="http://langgraph.test", + stream=True, + ) + streamed = [(chunk.choices[0].delta.content, chunk.choices[0].finish_reason) async for chunk in response] + + assert streamed == [("25 * 4", None), (" is 100", None), (None, "stop")] From 8bdce62ec26225eff7a94e4ade20811203535254 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:52:12 -0700 Subject: [PATCH 06/22] fix(streaming): keep the synthetic stop for a stream that sent no chunks Gemini's OpenAI-compatible endpoint answers with only [DONE] when reasoning uses up max_tokens. Failing every openai/azure stream that sent nothing turned that 200 into a 500, while every other provider already kept the stop for an empty stream. Adds integration cells for Gemini's trailing usage chunk, multi-event LangGraph runs, a mocked deployment and the [DONE]-only answer. --- .../litellm_core_utils/streaming_handler.py | 3 +- ..._stream_clean_eof_without_finish_reason.py | 94 ++++++++++++++++--- .../test_streaming_handler.py | 37 ++++++-- 3 files changed, 114 insertions(+), 20 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 70bad6395ef..d867f926d87 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -71,7 +71,6 @@ _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: @@ -1908,7 +1907,7 @@ class CustomStreamWrapper: def _should_fail_on_clean_eof(self) -> bool: return ( not self.is_mock - and (self.custom_llm_provider in _OPENAI_AZURE_CHAT_PROVIDERS or self.uses_openai_chat_stream) + and self.uses_openai_chat_stream and self.received_finish_reason is None and self.intermittent_finish_reason is None and not self.received_any_finish_reason diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index f1dad7e8844..eea9f8e449e 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -11,13 +11,19 @@ from integration._support.responses_stream import data_frames, spend_row, sse_ev from integration._support.wire import Reply, Request, wire_server from pydantic import JsonValue -Ending = Literal["ending-complete", "ending-clean-eof", "ending-done-without-finish"] -Family = Literal["openai", "anthropic", "hosted_vllm"] +Ending = Literal["ending-complete", "ending-clean-eof", "ending-done-without-finish", "ending-done-only"] +Family = Literal["openai", "anthropic", "hosted_vllm", "gemini", "langgraph"] -_ENDINGS: Final[tuple[Ending, ...]] = ("ending-complete", "ending-clean-eof", "ending-done-without-finish") +_ENDINGS: Final[tuple[Ending, ...]] = ( + "ending-complete", + "ending-clean-eof", + "ending-done-without-finish", + "ending-done-only", +) _PARTIAL: Final = ('{"findings":[{"title":', '"unfinished') _BACKEND: Final = "gpt-4o-mini" _CLAUDE: Final = "claude-sonnet-4-5" +_GEMINI: Final = "gemini-2.5-flash" _PROVIDER_KEY: Final = "synthetic-provider-key" _CALL_ID_HEADER: Final = "x-litellm-call-id" _USAGE: Final = {"prompt_tokens": 9, "completion_tokens": 7, "total_tokens": 16} @@ -25,15 +31,19 @@ _TARGETS: Final[Mapping[Family, str]] = { "openai": "/chat/completions", "hosted_vllm": "/chat/completions", "anthropic": "/v1/messages", + "gemini": f"/models/{_GEMINI}:streamGenerateContent", + "langgraph": "/runs/stream", } _DEPLOYMENTS: Final[Mapping[Family, str]] = { "openai": f"openai/{_BACKEND}", "hosted_vllm": f"hosted_vllm/{_BACKEND}", "anthropic": f"anthropic/{_CLAUDE}", + "gemini": f"gemini/{_GEMINI}", + "langgraph": "langgraph/agent", } -def _data(frame: Mapping[str, JsonValue]) -> bytes: +def _data(frame: JsonValue) -> bytes: return b"data: " + json.dumps(frame).encode() + b"\n\n" @@ -52,6 +62,8 @@ def _chat_frames(identity: str, ending: Ending) -> tuple[bytes, ...]: return content case "ending-done-without-finish": return (*content, b"data: [DONE]\n\n") + case "ending-done-only": + return (b"data: [DONE]\n\n",) def _anthropic_frames(identity: str, ending: Ending) -> tuple[bytes, ...]: @@ -59,19 +71,55 @@ def _anthropic_frames(identity: str, ending: Ending) -> tuple[bytes, ...]: return (*content, CONTENT_TAIL) if ending == "ending-complete" else content +def _gemini_frames(ending: Ending) -> tuple[bytes, ...]: + """Gemini puts finishReason on the last text chunk and can follow it with a usage-only chunk.""" + finished: Final = ending == "ending-complete" + content: Final = tuple( + _data( + { + "candidates": [ + { + "content": {"parts": [{"text": part}], "role": "model"}, + "index": 0, + **({"finishReason": "STOP"} if finished and part == _PARTIAL[-1] else {}), + } + ] + } + ) + for part in _PARTIAL + ) + usage: Final = _data({"usageMetadata": {"promptTokenCount": 9, "candidatesTokenCount": 7, "totalTokenCount": 16}}) + return (*content, usage) if finished else content + + +def _langgraph_frames() -> tuple[bytes, ...]: + """LangGraph ends a run by closing the stream after its metadata event.""" + return ( + *(_data(["messages", [[{"type": "ai", "content": part}, {}]]]) for part in _PARTIAL), + _data(["metadata", {"run_id": "run-1"}]), + ) + + +def _frames(family: Family, identity: str, ending: Ending) -> tuple[bytes, ...]: + match family: + case "anthropic": + return _anthropic_frames(identity, ending) + case "gemini": + return _gemini_frames(ending) + case "langgraph": + return _langgraph_frames() + case "openai" | "hosted_vllm": + return _chat_frames(identity, ending) + + def _ending(request: Request) -> Ending: return next(ending for ending in _ENDINGS if ending.encode() in request.body) def _upstream(family: Family) -> Callable[[Request], Reply]: def respond(request: Request) -> Reply: - assert request.method == "POST" and request.target.endswith(_TARGETS[family]), request.target - identity: Final = f"upstream-{uuid.uuid4().hex}" - frames: Final = ( - _anthropic_frames(identity, _ending(request)) - if family == "anthropic" - else _chat_frames(identity, _ending(request)) - ) + assert request.method == "POST" and _TARGETS[family] in request.target, request.target + frames: Final = _frames(family, f"upstream-{uuid.uuid4().hex}", _ending(request)) return Reply(content_type="text/event-stream", chunks=frames) return answering_model_discovery(respond) @@ -150,7 +198,18 @@ def test_chat_stream_that_sends_done_without_a_finish_reason_reaches_the_client_ assert spend_row(call_id)["status"] == "failure" -@pytest.mark.parametrize("family", ["openai", "anthropic"]) +def test_chat_stream_that_sends_only_done_still_completes_empty(gateway: Gateway) -> None: + """Gemini's OpenAI-compatible endpoint answers with only [DONE] when reasoning uses up max_tokens.""" + with wire_server(_upstream("openai")) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=_DEPLOYMENTS["openai"], api_base=wire.url, api_key=_PROVIDER_KEY) + text, call_id = _chat(gateway, model, "ending-done-only") + assert _chat_text(text) == "", text + assert _finish_reasons(text) == ("stop",), text + assert _chat_errors(text) == (), text + assert spend_row(call_id)["status"] == "success" + + +@pytest.mark.parametrize("family", ["openai", "anthropic", "gemini", "langgraph"]) def test_chat_stream_with_a_provider_finish_reason_still_completes(gateway: Gateway, family: Family) -> None: with wire_server(_upstream(family)) as wire, gateway.scenario() as scenario: model: Final = scenario.model(model=_DEPLOYMENTS[family], api_base=wire.url, api_key=_PROVIDER_KEY) @@ -162,6 +221,17 @@ def test_chat_stream_with_a_provider_finish_reason_still_completes(gateway: Gate assert spend_row(call_id)["status"] == "success" +def test_chat_stream_from_a_mock_response_deployment_still_completes(gateway: Gateway) -> None: + """mock_completion streams never carry a finish reason, so a mocked deployment must keep its synthetic stop.""" + with gateway.scenario() as scenario: + model: Final = scenario.model(model=_DEPLOYMENTS["openai"], mock_response="".join(_PARTIAL)) + text, call_id = _chat(gateway, model, "ending-clean-eof") + assert _chat_text(text) == "".join(_PARTIAL), text + assert _finish_reasons(text) == ("stop",), text + assert _chat_errors(text) == (), text + assert spend_row(call_id)["status"] == "success" + + @pytest.mark.parametrize(("ending", "terminal"), [("ending-complete", True), ("ending-clean-eof", False)]) def test_messages_over_a_chat_bridged_deployment_reports_end_turn_only_for_a_finished_stream( gateway: Gateway, ending: Ending, terminal: bool diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 2d1c2a581cf..0586af6e72b 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -1651,6 +1651,37 @@ def test_sync_openai_compatible_chat_clean_eof_without_finish_reason_raises_mids assert response.received_finish_reason is None +async def _collect(wrapper: CustomStreamWrapper, sync_mode: bool) -> list[ModelResponseStream]: + if sync_mode: + return list(wrapper) + return [chunk async for chunk in wrapper] + + +@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure", "hosted_vllm"]) +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_stream_that_sends_no_chunks_keeps_the_default_stop( + logging_obj: Logging, custom_llm_provider: str, sync_mode: bool +): + """Gemini's OpenAI-compatible endpoint answers with only [DONE] when reasoning uses up max_tokens.""" + + async def no_chunks() -> AsyncIterator[ModelResponseStream]: + for chunk in (): + yield chunk + + wrapper: Final = CustomStreamWrapper( + completion_stream=iter(()) if sync_mode else no_chunks(), + model="gemini-2.5-flash", + logging_obj=logging_obj, + custom_llm_provider=custom_llm_provider, + stream_options={"include_usage": True}, + ) + + received: Final = await _collect(wrapper, sync_mode) + + assert [c.choices[0].finish_reason for c in received if c.choices and c.choices[0].finish_reason] == ["stop"] + + _ANTHROPIC_PARTIAL_EVENTS: Final[tuple[dict[str, object], ...]] = ( { "type": "message_start", @@ -1701,12 +1732,6 @@ def _anthropic_stream( ) -async def _collect(wrapper: CustomStreamWrapper, sync_mode: bool) -> list[ModelResponseStream]: - if sync_mode: - return list(wrapper) - return [chunk async for chunk in wrapper] - - @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio async def test_anthropic_stream_that_ends_before_message_delta_fails_with_the_eof_message( From aec3fbaeca895e820848a88437a60f0d320c8521 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 14:08:07 -0700 Subject: [PATCH 07/22] test(streaming): keep pytest.raises to one statement, drop the spend read for an empty stream An empty stream writes no spend row on main either, so the [DONE]-only cell checks only what the client sees. --- .../test_stream_clean_eof_without_finish_reason.py | 4 ++-- tests/unit/litellm_core_utils/test_streaming_handler.py | 6 +----- 2 files changed, 3 insertions(+), 7 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index eea9f8e449e..7588c255a23 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -202,11 +202,11 @@ def test_chat_stream_that_sends_only_done_still_completes_empty(gateway: Gateway """Gemini's OpenAI-compatible endpoint answers with only [DONE] when reasoning uses up max_tokens.""" with wire_server(_upstream("openai")) as wire, gateway.scenario() as scenario: model: Final = scenario.model(model=_DEPLOYMENTS["openai"], api_base=wire.url, api_key=_PROVIDER_KEY) - text, call_id = _chat(gateway, model, "ending-done-only") + text, _ = _chat(gateway, model, "ending-done-only") assert _chat_text(text) == "", text assert _finish_reasons(text) == ("stop",), text assert _chat_errors(text) == (), text - assert spend_row(call_id)["status"] == "success" + assert text.rstrip().endswith("data: [DONE]"), text @pytest.mark.parametrize("family", ["openai", "anthropic", "gemini", "langgraph"]) diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 0586af6e72b..5043cccb053 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -1577,7 +1577,6 @@ async def test_mid_stream_error_keeps_credentials_in_the_api_base_out(routed_api assert "deployment-secret" not in debug_info - def _incomplete_openai_chat_chunk() -> ModelResponseStream: return ModelResponseStream( id="chatcmpl-incomplete-eof", @@ -7686,10 +7685,7 @@ async def test_no_synthetic_finish_reason_logged_when_provider_sent_none(sync_mo logging_obj=logging_obj, ) with pytest.raises(MidStreamFallbackError): - if sync_mode: - list(response) - else: - [c async for c in response] + await _collect(response, sync_mode) assert response.received_finish_reason is None assert all(not (c.choices and c.choices[0].finish_reason) for c in response.chunks) From 04f3765e1afd9236de5205678ff5c40169d86388 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 14:47:29 -0700 Subject: [PATCH 08/22] fix(streaming): read the chunk's finish reason for the sticky flag --- litellm/litellm_core_utils/streaming_handler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index d867f926d87..ce06f744e6f 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -1534,7 +1534,7 @@ class CustomStreamWrapper: return _ProviderChunkEarlyReturn(None) completion_obj["content"] = response_obj["text"] self.intermittent_finish_reason = response_obj.get("finish_reason", None) - self.received_any_finish_reason |= self.intermittent_finish_reason is not None + self.received_any_finish_reason |= response_obj["finish_reason"] is not None if response_obj["is_finished"]: if response_obj["finish_reason"] == "error": raise Exception( From 0ea2557ffe0871d25835fc44d4df075a331dceb0 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:00:11 -0700 Subject: [PATCH 09/22] fix(agentcore): end a stream on messageStop when the runtime sends no metadata or final message --- .../bedrock/chat/agentcore/transformation.py | 27 +++++++ .../test_agentcore_transformation.py | 80 ++++++++++++++++++- 2 files changed, 106 insertions(+), 1 deletion(-) diff --git a/litellm/llms/bedrock/chat/agentcore/transformation.py b/litellm/llms/bedrock/chat/agentcore/transformation.py index 48b3795e274..b3706b6c4e8 100644 --- a/litellm/llms/bedrock/chat/agentcore/transformation.py +++ b/litellm/llms/bedrock/chat/agentcore/transformation.py @@ -540,6 +540,17 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): return AgentCoreParsedResponse(content=content, usage=usage_data, final_message=final_message) + @staticmethod + def _message_stop_chunk(model: str) -> ModelResponseStream: + chunk: Final = ModelResponseStream( + id=f"chatcmpl-{uuid.uuid4()}", + created=0, + model=model, + object="chat.completion.chunk", + ) + chunk.choices = [StreamingChoices(finish_reason="stop", index=0, delta=Delta())] + return chunk + def _stream_agentcore_response_sync( self, response: httpx.Response, @@ -549,6 +560,8 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): Internal sync generator that parses SSE and yields ModelResponse chunks. """ buffer = "" + sent_finish = False + message_stopped = False # runtimes that send no metadata or final message end on messageStop for text_chunk in response.iter_text(): buffer += text_chunk @@ -620,7 +633,9 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): total_tokens=usage_data.get("totalTokens", 0), ), ) + sent_finish = True yield chunk + message_stopped |= "messageStop" in event_payload # Process final message if "message" in data_obj and isinstance(data_obj["message"], dict): @@ -637,12 +652,16 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): delta=Delta(), ) ] + sent_finish = True yield chunk except json.JSONDecodeError: verbose_logger.debug("Skipping non-JSON SSE line: %s", line[:100]) continue + if message_stopped and not sent_finish: + yield self._message_stop_chunk(model) + def get_sync_custom_stream_wrapper( self, model: str, @@ -771,6 +790,8 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): Internal async generator that parses SSE and yields ModelResponse chunks. """ buffer = "" + sent_finish = False + message_stopped = False # runtimes that send no metadata or final message end on messageStop async for text_chunk in response.aiter_text(): buffer += text_chunk @@ -842,7 +863,9 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): total_tokens=usage_data.get("totalTokens", 0), ), ) + sent_finish = True yield chunk + message_stopped |= "messageStop" in event_payload # Process final message if "message" in data_obj and isinstance(data_obj["message"], dict): @@ -859,12 +882,16 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): delta=Delta(), ) ] + sent_finish = True yield chunk except json.JSONDecodeError: verbose_logger.debug("Skipping non-JSON SSE line: %s", line[:100]) continue + if message_stopped and not sent_finish: + yield self._message_stop_chunk(model) + async def get_async_custom_stream_wrapper( self, model: str, diff --git a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py b/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py index 053ba477212..8ad70d120ac 100644 --- a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py +++ b/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py @@ -19,8 +19,9 @@ from typing import Final from unittest.mock import MagicMock, Mock, patch import litellm +from litellm.exceptions import MidStreamFallbackError from litellm.llms.bedrock.chat.agentcore.transformation import AmazonAgentCoreConfig -from litellm.llms.custom_httpx.http_handler import HTTPHandler +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler class TestAgentCoreAcceptHeader: @@ -1269,3 +1270,80 @@ def test_agentcore_runtime_response_becomes_assistant_message( assert json.loads(route.calls[0].request.content)["prompt"] == "Explain machine learning in simple terms" assert response.choices[0].message.content == "Machine learning lets computers learn from data." assert response.choices[0].finish_reason == "stop" + + +_AGENTCORE_RUNTIME_URL: Final = ( + "https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/" + "arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A888602223428%3Aruntime%2Fhosted_agent_13sf6-cALnp38iZD/invocations" +) +_AGENTCORE_DELTAS: Final = ( + 'data: {"event":{"contentBlockDelta":{"delta":{"text":"Machine learning lets "}}}}\n\n' + 'data: {"event":{"contentBlockDelta":{"delta":{"text":"computers learn from data."}}}}\n\n' +) + + +async def _stream_agentcore(body: str, sync_mode: bool) -> list[tuple[str | None, str | None]]: + """Stream `body` from the runtime as SSE and return each chunk's (content, finish_reason).""" + kwargs: Final = { + "model": "bedrock/agentcore/arn:aws:bedrock-agentcore:us-west-2:888602223428:runtime/" + "hosted_agent_13sf6-cALnp38iZD", + "messages": [{"role": "user", "content": "Explain machine learning in simple terms"}], + "aws_access_key_id": "AKIAAGENTCOREKEY", + "aws_secret_access_key": "agentcore-secret", + "stream": True, + } + if sync_mode: + return [ + (chunk.choices[0].delta.content, chunk.choices[0].finish_reason) + for chunk in litellm.completion(**kwargs, client=HTTPHandler()) + ] + return [ + (chunk.choices[0].delta.content, chunk.choices[0].finish_reason) + async for chunk in await litellm.acompletion(**kwargs, client=AsyncHTTPHandler()) + ] + + +@pytest.mark.parametrize( + "ending", + [ + pytest.param('data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n', id="message_stop"), + pytest.param( + 'data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n' + 'data: {"event":{"metadata":{"usage":{"inputTokens":9,"outputTokens":7,"totalTokens":16}}}}\n\n' + 'data: {"message":{"role":"assistant","content":[{"text":"Machine learning lets computers learn from data."}]}}\n\n', + id="strands_full", + ), + ], +) +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.respx(assert_all_called=True) +async def test_agentcore_stream_that_signals_its_end_completes( + ending: str, sync_mode: bool, respx_mock: respx.Router, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + respx_mock.post(_AGENTCORE_RUNTIME_URL).mock( + return_value=httpx.Response( + 200, content=(_AGENTCORE_DELTAS + ending).encode(), headers={"content-type": "text/event-stream"} + ) + ) + + chunks: Final = await _stream_agentcore(_AGENTCORE_DELTAS + ending, sync_mode) + + assert "".join(content for content, _ in chunks if content) == "Machine learning lets computers learn from data." + assert [finish for _, finish in chunks if finish] == ["stop"] + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.respx(assert_all_called=True) +async def test_agentcore_stream_cut_before_it_signals_its_end_raises( + sync_mode: bool, respx_mock: respx.Router, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + respx_mock.post(_AGENTCORE_RUNTIME_URL).mock( + return_value=httpx.Response( + 200, content=_AGENTCORE_DELTAS.encode(), headers={"content-type": "text/event-stream"} + ) + ) + + with pytest.raises(MidStreamFallbackError, match="Stream ended before a finish_reason was received"): + await _stream_agentcore(_AGENTCORE_DELTAS, sync_mode) From 4a9a2347424953ce0e923a3f8e1c33115843f1f6 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:00:11 -0700 Subject: [PATCH 10/22] test(streaming): pin the logged-chunks guard, inject the LangGraph client, use current model names --- ..._stream_clean_eof_without_finish_reason.py | 20 ++++++----- .../test_streaming_handler.py | 34 +++++-------------- .../chat/test_langgraph_transformation.py | 12 +++---- 3 files changed, 24 insertions(+), 42 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index 7588c255a23..fa1125940da 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -21,9 +21,9 @@ _ENDINGS: Final[tuple[Ending, ...]] = ( "ending-done-only", ) _PARTIAL: Final = ('{"findings":[{"title":', '"unfinished') -_BACKEND: Final = "gpt-4o-mini" -_CLAUDE: Final = "claude-sonnet-4-5" -_GEMINI: Final = "gemini-2.5-flash" +_BACKEND: Final = "gpt-5.6" +_CLAUDE: Final = "claude-sonnet-5-5" +_GEMINI: Final = "gemini-3.8-flash" _PROVIDER_KEY: Final = "synthetic-provider-key" _CALL_ID_HEADER: Final = "x-litellm-call-id" _USAGE: Final = {"prompt_tokens": 9, "completion_tokens": 7, "total_tokens": 16} @@ -113,7 +113,9 @@ def _frames(family: Family, identity: str, ending: Ending) -> tuple[bytes, ...]: def _ending(request: Request) -> Ending: - return next(ending for ending in _ENDINGS if ending.encode() in request.body) + endings: Final = [ending for ending in _ENDINGS if ending.encode() in request.body] + assert len(endings) == 1, request.body + return endings[0] def _upstream(family: Family) -> Callable[[Request], Reply]: @@ -157,14 +159,13 @@ def _finish_reasons(text: str) -> tuple[str, ...]: def _chat_text(text: str) -> str: - return "".join( - string_value(delta.get("content") or "") + deltas: Final = ( + choice.get("delta") for frame in data_frames(text) for choice in frame.get("choices") or [] if isinstance(choice, dict) - for delta in (choice.get("delta") or {},) - if isinstance(delta, dict) ) + return "".join(string_value(delta.get("content") or "") for delta in deltas if isinstance(delta, dict)) def _chat_errors(text: str) -> tuple[Mapping[str, JsonValue], ...]: @@ -199,7 +200,8 @@ def test_chat_stream_that_sends_done_without_a_finish_reason_reaches_the_client_ def test_chat_stream_that_sends_only_done_still_completes_empty(gateway: Gateway) -> None: - """Gemini's OpenAI-compatible endpoint answers with only [DONE] when reasoning uses up max_tokens.""" + """An upstream can answer with only [DONE]: Gemini's OpenAI-compatible endpoint did on 2026-10-20 when + reasoning used up max_tokens.""" with wire_server(_upstream("openai")) as wire, gateway.scenario() as scenario: model: Final = scenario.model(model=_DEPLOYMENTS["openai"], api_base=wire.url, api_key=_PROVIDER_KEY) text, _ = _chat(gateway, model, "ending-done-only") diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 5043cccb053..b44f7a5c7a7 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -1580,7 +1580,7 @@ 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()), + created=1, model="gpt-5.6", object="chat.completion.chunk", choices=[ @@ -1600,12 +1600,9 @@ async def test_async_openai_compatible_chat_clean_eof_without_finish_reason_rais ): """An OpenAI-compatible stream without a provider finish reason must fail at clean EOF.""" - async def source(): - yield _incomplete_openai_chat_chunk() - logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider response = CustomStreamWrapper( - completion_stream=source(), + completion_stream=_as_async(iter([_incomplete_openai_chat_chunk()])), model="gpt-5.6", logging_obj=logging_obj, custom_llm_provider=custom_llm_provider, @@ -1628,12 +1625,9 @@ def test_sync_openai_compatible_chat_clean_eof_without_finish_reason_raises_mids ): """The synchronous OpenAI-compatible stream must fail at clean EOF too.""" - def source(): - yield _incomplete_openai_chat_chunk() - logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider response = CustomStreamWrapper( - completion_stream=source(), + completion_stream=iter([_incomplete_openai_chat_chunk()]), model="gpt-5.6", logging_obj=logging_obj, custom_llm_provider=custom_llm_provider, @@ -1664,12 +1658,8 @@ async def test_stream_that_sends_no_chunks_keeps_the_default_stop( ): """Gemini's OpenAI-compatible endpoint answers with only [DONE] when reasoning uses up max_tokens.""" - async def no_chunks() -> AsyncIterator[ModelResponseStream]: - for chunk in (): - yield chunk - wrapper: Final = CustomStreamWrapper( - completion_stream=iter(()) if sync_mode else no_chunks(), + completion_stream=iter(()) if sync_mode else _as_async(iter(())), model="gemini-2.5-flash", logging_obj=logging_obj, custom_llm_provider=custom_llm_provider, @@ -1717,13 +1707,9 @@ def _anthropic_stream( ) -> CustomStreamWrapper: lines: Final = _anthropic_sse_lines(events) - async def async_lines() -> AsyncIterator[str]: - for line in lines: - yield line - return CustomStreamWrapper( completion_stream=AnthropicModelResponseIterator( - streaming_response=iter(lines) if sync_mode else async_lines(), sync_stream=sync_mode + streaming_response=iter(lines) if sync_mode else _as_async(iter(lines)), sync_stream=sync_mode ), model="claude-sonnet-4-5", custom_llm_provider="anthropic", @@ -1768,7 +1754,8 @@ async def test_anthropic_stream_with_message_stop_ends_with_its_finish_reason(sy async def test_generic_chunk_stream_without_a_finish_reason_keeps_the_default_stop( sync_mode: bool, logging_obj: Logging ): - """Providers that stream GenericStreamingChunks and never mark the last one finished still end with stop.""" + """Providers that stream GenericStreamingChunks and never mark the last one finished still end with stop, + but the logged response built from ``chunks`` never gains that synthetic stop.""" chunks: Final = [ {"text": text, "tool_use": None, "is_finished": False, "finish_reason": "", "usage": None, "index": 0} for text in ("partial", " reply") @@ -1784,6 +1771,7 @@ async def test_generic_chunk_stream_without_a_finish_reason_keeps_the_default_st assert [c.choices[0].finish_reason for c in received if c.choices and c.choices[0].finish_reason] == ["stop"] assert "".join(c.choices[0].delta.content or "" for c in received if c.choices) == "partial reply" + assert all(not (c.choices and c.choices[0].finish_reason) for c in wrapper.chunks) @pytest.mark.parametrize("sync_mode", [True, False]) @@ -1805,15 +1793,11 @@ async def test_gemini_stream_with_a_usage_chunk_after_its_finish_reason_still_co ) ] - async def async_lines() -> AsyncIterator[str]: - for line in lines: - yield line - logging_obj.update_environment_variables( model="gemini-2.5-flash", user=None, optional_params={}, litellm_params={"litellm_call_id": "gemini-tail"} ) iterator: Final = GeminiModelResponseIterator( - streaming_response=iter(lines) if sync_mode else async_lines(), sync_stream=sync_mode, logging_obj=logging_obj + streaming_response=iter(lines) if sync_mode else _as_async(iter(lines)), sync_stream=sync_mode, logging_obj=logging_obj ) wrapper: Final = CustomStreamWrapper( completion_stream=iter(iterator) if sync_mode else iterator, diff --git a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py index 6612c71f436..4db85bc8f58 100644 --- a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py +++ b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py @@ -27,7 +27,7 @@ import httpx import pytest import litellm -from litellm.llms.custom_httpx.http_handler import get_async_httpx_client +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, get_async_httpx_client from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome @@ -271,24 +271,20 @@ async def test_langgraph_streaming_acompletion_without_a_client_posts_through_th @pytest.mark.asyncio -async def test_langgraph_async_stream_delivers_every_message_event_before_its_stop(monkeypatch): +async def test_langgraph_async_stream_delivers_every_message_event_before_its_stop(): async def events(): yield b'data: ["messages", [[{"type": "ai", "content": "25 * 4"}, {}]]]\n\n' yield b'data: ["messages", [[{"type": "ai", "content": " is 100"}, {}]]]\n\n' yield b'data: ["metadata", {"run_id": "run-1"}]\n\n' - cached_client = get_async_httpx_client(llm_provider="langgraph", params={}) - monkeypatch.setattr( - cached_client, - "client", - httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200, content=events()))), - ) + client = AsyncHTTPHandler(transport=httpx.MockTransport(lambda request: httpx.Response(200, content=events()))) response = await litellm.acompletion( model="langgraph/agent", messages=[{"role": "user", "content": "What is 25 * 4?"}], api_base="http://langgraph.test", stream=True, + client=client, ) streamed = [(chunk.choices[0].delta.content, chunk.choices[0].finish_reason) async for chunk in response] From b3efc0fcf2b300f7397477f3e598f87d447acd1b Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:03:22 -0700 Subject: [PATCH 11/22] fix(agentcore): say why the end-of-stream flags rebind --- litellm/llms/bedrock/chat/agentcore/transformation.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/litellm/llms/bedrock/chat/agentcore/transformation.py b/litellm/llms/bedrock/chat/agentcore/transformation.py index b3706b6c4e8..afe7f8872fe 100644 --- a/litellm/llms/bedrock/chat/agentcore/transformation.py +++ b/litellm/llms/bedrock/chat/agentcore/transformation.py @@ -560,8 +560,8 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): Internal sync generator that parses SSE and yields ModelResponse chunks. """ buffer = "" - sent_finish = False - message_stopped = False # runtimes that send no metadata or final message end on messageStop + sent_finish = False # rebind-ok: flips when a finish chunk goes out + message_stopped = False # rebind-ok: flips on messageStop, the only end some runtimes send for text_chunk in response.iter_text(): buffer += text_chunk @@ -790,8 +790,8 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM): Internal async generator that parses SSE and yields ModelResponse chunks. """ buffer = "" - sent_finish = False - message_stopped = False # runtimes that send no metadata or final message end on messageStop + sent_finish = False # rebind-ok: flips when a finish chunk goes out + message_stopped = False # rebind-ok: flips on messageStop, the only end some runtimes send async for text_chunk in response.aiter_text(): buffer += text_chunk From 68f27a760dc958a15f67e8f980053af416eb40a2 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:06:18 -0700 Subject: [PATCH 12/22] test(streaming): cover an AgentCore runtime that ends on messageStop through the proxy --- ..._stream_clean_eof_without_finish_reason.py | 38 +++++++++++++++---- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index fa1125940da..a23c50b3b02 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -5,14 +5,14 @@ from typing import Final, Literal import pytest from integration._support.anthropic_sse import CONTENT_BLOCK_START, CONTENT_TAIL, message_start, text_delta -from integration._support.client import Gateway, string_value +from integration._support.client import Gateway, Scenario, string_value from integration._support.openai_wire import answering_model_discovery from integration._support.responses_stream import data_frames, spend_row, sse_events -from integration._support.wire import Reply, Request, wire_server +from integration._support.wire import Reply, Request, Wire, wire_server from pydantic import JsonValue Ending = Literal["ending-complete", "ending-clean-eof", "ending-done-without-finish", "ending-done-only"] -Family = Literal["openai", "anthropic", "hosted_vllm", "gemini", "langgraph"] +Family = Literal["openai", "anthropic", "hosted_vllm", "gemini", "langgraph", "agentcore"] _ENDINGS: Final[tuple[Ending, ...]] = ( "ending-complete", @@ -24,6 +24,7 @@ _PARTIAL: Final = ('{"findings":[{"title":', '"unfinished') _BACKEND: Final = "gpt-5.6" _CLAUDE: Final = "claude-sonnet-5-5" _GEMINI: Final = "gemini-3.8-flash" +_AGENT_RUNTIME: Final = "arn:aws:bedrock-agentcore:us-west-2:888602223428:runtime/integration-agent" _PROVIDER_KEY: Final = "synthetic-provider-key" _CALL_ID_HEADER: Final = "x-litellm-call-id" _USAGE: Final = {"prompt_tokens": 9, "completion_tokens": 7, "total_tokens": 16} @@ -33,6 +34,7 @@ _TARGETS: Final[Mapping[Family, str]] = { "anthropic": "/v1/messages", "gemini": f"/models/{_GEMINI}:streamGenerateContent", "langgraph": "/runs/stream", + "agentcore": "/runtimes/", } _DEPLOYMENTS: Final[Mapping[Family, str]] = { "openai": f"openai/{_BACKEND}", @@ -40,6 +42,7 @@ _DEPLOYMENTS: Final[Mapping[Family, str]] = { "anthropic": f"anthropic/{_CLAUDE}", "gemini": f"gemini/{_GEMINI}", "langgraph": "langgraph/agent", + "agentcore": f"bedrock/agentcore/{_AGENT_RUNTIME}", } @@ -100,8 +103,17 @@ def _langgraph_frames() -> tuple[bytes, ...]: ) +def _agentcore_frames(ending: Ending) -> tuple[bytes, ...]: + """A runtime that is not Strands can end its turn on messageStop with no metadata or final message.""" + content: Final = tuple(_data({"event": {"contentBlockDelta": {"delta": {"text": part}}}}) for part in _PARTIAL) + message_stop: Final = _data({"event": {"messageStop": {"stopReason": "end_turn"}}}) + return (*content, message_stop) if ending == "ending-complete" else content + + def _frames(family: Family, identity: str, ending: Ending) -> tuple[bytes, ...]: match family: + case "agentcore": + return _agentcore_frames(ending) case "anthropic": return _anthropic_frames(identity, ending) case "gemini": @@ -127,6 +139,18 @@ def _upstream(family: Family) -> Callable[[Request], Reply]: return answering_model_discovery(respond) +def _deploy(scenario: Scenario, family: Family, wire: Wire) -> str: + if family == "agentcore": + return scenario.model( + model=_DEPLOYMENTS[family], + aws_bedrock_runtime_endpoint=wire.url, + aws_access_key_id="AKIAINTEGRATIONAGENT", + aws_secret_access_key="synthetic-agentcore-secret", + aws_region_name="us-west-2", + ) + return scenario.model(model=_DEPLOYMENTS[family], api_base=wire.url, api_key=_PROVIDER_KEY) + + def _prompt(ending: Ending) -> str: return f"Return JSON {ending} {uuid.uuid4().hex}" @@ -176,12 +200,12 @@ def _event_types(text: str) -> tuple[str, ...]: return tuple(event for event, _ in sse_events(text)) -@pytest.mark.parametrize("family", ["openai", "anthropic"]) +@pytest.mark.parametrize("family", ["openai", "anthropic", "agentcore"]) def test_chat_stream_that_ends_cleanly_without_a_finish_reason_reaches_the_client_as_an_error( gateway: Gateway, family: Family ) -> None: with wire_server(_upstream(family)) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=_DEPLOYMENTS[family], api_base=wire.url, api_key=_PROVIDER_KEY) + model: Final = _deploy(scenario, family, wire) text, call_id = _chat(gateway, model, "ending-clean-eof") assert _chat_text(text) == "".join(_PARTIAL), text assert _finish_reasons(text) == (), text @@ -211,10 +235,10 @@ def test_chat_stream_that_sends_only_done_still_completes_empty(gateway: Gateway assert text.rstrip().endswith("data: [DONE]"), text -@pytest.mark.parametrize("family", ["openai", "anthropic", "gemini", "langgraph"]) +@pytest.mark.parametrize("family", ["openai", "anthropic", "gemini", "langgraph", "agentcore"]) def test_chat_stream_with_a_provider_finish_reason_still_completes(gateway: Gateway, family: Family) -> None: with wire_server(_upstream(family)) as wire, gateway.scenario() as scenario: - model: Final = scenario.model(model=_DEPLOYMENTS[family], api_base=wire.url, api_key=_PROVIDER_KEY) + model: Final = _deploy(scenario, family, wire) text, call_id = _chat(gateway, model, "ending-complete") assert _chat_text(text) == "".join(_PARTIAL), text assert _finish_reasons(text) == ("stop",), text From 793e6d849bdc8fac3c0d62c3cdd4a3799fc34a92 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:44:26 -0700 Subject: [PATCH 13/22] test(streaming): cover SDK clients, request fallbacks and a concurrent burst for a cut stream --- ..._stream_clean_eof_without_finish_reason.py | 182 ++++++++++++++++-- 1 file changed, 166 insertions(+), 16 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index a23c50b3b02..024783aabfb 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -1,14 +1,26 @@ +import asyncio +import itertools import json import uuid -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterator, Mapping +from concurrent.futures import ThreadPoolExecutor +from contextlib import ExitStack from typing import Final, Literal +import anthropic +import openai import pytest from integration._support.anthropic_sse import CONTENT_BLOCK_START, CONTENT_TAIL, message_start, text_delta from integration._support.client import Gateway, Scenario, string_value -from integration._support.openai_wire import answering_model_discovery +from integration._support.openai_wire import answering_model_discovery, posted_targets from integration._support.responses_stream import data_frames, spend_row, sse_events from integration._support.wire import Reply, Request, Wire, wire_server +from integration.providers._count_tokens_system_lift import ( + anthropic_client, + async_anthropic_client, + async_openai_client, + openai_client, +) from pydantic import JsonValue Ending = Literal["ending-complete", "ending-clean-eof", "ending-done-without-finish", "ending-done-only"] @@ -130,10 +142,10 @@ def _ending(request: Request) -> Ending: return endings[0] -def _upstream(family: Family) -> Callable[[Request], Reply]: +def _upstream(family: Family, ending: Ending | None = None) -> Callable[[Request], Reply]: def respond(request: Request) -> Reply: assert request.method == "POST" and _TARGETS[family] in request.target, request.target - frames: Final = _frames(family, f"upstream-{uuid.uuid4().hex}", _ending(request)) + frames: Final = _frames(family, f"upstream-{uuid.uuid4().hex}", ending or _ending(request)) return Reply(content_type="text/event-stream", chunks=frames) return answering_model_discovery(respond) @@ -173,22 +185,17 @@ def _chat(gateway: Gateway, model: str, ending: Ending) -> tuple[str, str]: ) +def _choices(text: str) -> Iterator[Mapping[str, JsonValue]]: + for frame in data_frames(text): + yield from (choice for choice in frame.get("choices") or [] if isinstance(choice, dict)) + + def _finish_reasons(text: str) -> tuple[str, ...]: - return tuple( - string_value(choice["finish_reason"]) - for frame in data_frames(text) - for choice in frame.get("choices") or [] - if isinstance(choice, dict) and choice.get("finish_reason") - ) + return tuple(string_value(choice["finish_reason"]) for choice in _choices(text) if choice.get("finish_reason")) def _chat_text(text: str) -> str: - deltas: Final = ( - choice.get("delta") - for frame in data_frames(text) - for choice in frame.get("choices") or [] - if isinstance(choice, dict) - ) + deltas: Final = (choice.get("delta") for choice in _choices(text)) return "".join(string_value(delta.get("content") or "") for delta in deltas if isinstance(delta, dict)) @@ -294,3 +301,146 @@ def test_responses_over_a_chat_bridged_anthropic_deployment_completes_only_a_fin assert ("response.completed" in types) is terminal, text assert (types[-1] in {"error", "response.failed"}) is not terminal, text assert spend_row(call_id)["status"] == ("success" if terminal else "failure") + + +def _drain_openai_sdk(gateway: Gateway, model: str, sync_mode: bool) -> None: + messages: Final = [{"role": "user", "content": _prompt("ending-clean-eof")}] + if sync_mode: + for _ in openai_client(gateway).chat.completions.create(model=model, messages=messages, stream=True): + pass + return + + async def drain() -> None: + stream: Final = await async_openai_client(gateway).chat.completions.create( + model=model, messages=messages, stream=True + ) + async for _ in stream: + pass + + asyncio.run(drain()) + + +def _drain_anthropic_sdk(gateway: Gateway, model: str, sync_mode: bool) -> None: + messages: Final = [{"role": "user", "content": _prompt("ending-clean-eof")}] + if sync_mode: + for _ in anthropic_client(gateway).messages.create(model=model, max_tokens=64, messages=messages, stream=True): + pass + return + + async def drain() -> None: + stream: Final = await async_anthropic_client(gateway).messages.create( + model=model, max_tokens=64, messages=messages, stream=True + ) + async for _ in stream: + pass + + asyncio.run(drain()) + + +@pytest.mark.parametrize("sync_mode", [True, False]) +def test_openai_sdk_stream_cut_before_a_finish_reason_raises(gateway: Gateway, sync_mode: bool) -> None: + with wire_server(_upstream("openai")) as wire, gateway.scenario() as scenario: + model: Final = _deploy(scenario, "openai", wire) + with pytest.raises(openai.APIError, match="Stream ended before a finish_reason was received"): + _drain_openai_sdk(gateway, model, sync_mode) + + +@pytest.mark.parametrize("sync_mode", [True, False]) +def test_anthropic_sdk_messages_stream_cut_before_a_finish_reason_raises(gateway: Gateway, sync_mode: bool) -> None: + with wire_server(_upstream("hosted_vllm")) as wire, gateway.scenario() as scenario: + model: Final = _deploy(scenario, "hosted_vllm", wire) + with pytest.raises(anthropic.APIError, match="Stream ended before a finish_reason was received"): + _drain_anthropic_sdk(gateway, model, sync_mode) + + +def test_responses_stream_cut_on_its_deployment_finishes_on_the_request_fallback(gateway: Gateway) -> None: + with ( + wire_server(_upstream("anthropic")) as primary, + wire_server(_upstream("anthropic", "ending-complete")) as backup, + gateway.scenario() as scenario, + ): + model: Final = _deploy(scenario, "anthropic", primary) + fallback: Final = _deploy(scenario, "anthropic", backup) + text, _ = _post( + gateway, "/v1/responses", {"model": model, "input": _prompt("ending-clean-eof"), "fallbacks": [fallback]} + ) + types: Final = tuple(string_value(frame.get("type") or "") for frame in data_frames(text)) + assert "response.completed" in types and "response.failed" not in types, text + assert (len(posted_targets(primary)), len(posted_targets(backup))) == (1, 1), text + + +def test_chat_stream_cut_after_its_content_is_not_replayed_on_the_request_fallback(gateway: Gateway) -> None: + with ( + wire_server(_upstream("openai")) as primary, + wire_server(_upstream("openai", "ending-complete")) as backup, + gateway.scenario() as scenario, + ): + model: Final = _deploy(scenario, "openai", primary) + fallback: Final = _deploy(scenario, "openai", backup) + text, _ = _post( + gateway, + "/v1/chat/completions", + { + "model": model, + "messages": [{"role": "user", "content": _prompt("ending-clean-eof")}], + "fallbacks": [fallback], + }, + ) + assert _chat_text(text) == "".join(_PARTIAL), text + assert len(_chat_errors(text)) == 1, text + assert posted_targets(backup) == (), text + + +Route = Literal["/v1/chat/completions", "/v1/messages", "/v1/responses"] + + +def _route_body(route: Route, model: str, ending: Ending) -> Mapping[str, JsonValue]: + match route: + case "/v1/chat/completions": + return {"model": model, "messages": [{"role": "user", "content": _prompt(ending)}]} + case "/v1/messages": + return {"model": model, "max_tokens": 64, "messages": [{"role": "user", "content": _prompt(ending)}]} + case "/v1/responses": + return {"model": model, "input": _prompt(ending)} + + +def _finished(route: Route, text: str) -> bool: + match route: + case "/v1/chat/completions": + return _finish_reasons(text) == ("stop",) and _chat_errors(text) == () + case "/v1/messages": + return "message_stop" in _event_types(text) and "error" not in _event_types(text) + case "/v1/responses": + return "response.completed" in tuple(string_value(frame.get("type") or "") for frame in data_frames(text)) + + +def _stream_outcome(gateway: Gateway, model: str, route: Route, ending: Ending) -> tuple[bool, str]: + text, call_id = _post(gateway, route, _route_body(route, model, ending)) + return _finished(route, text), call_id + + +_BRIDGED_FAMILY: Final[Mapping[Route, Family]] = { + "/v1/chat/completions": "openai", + "/v1/messages": "hosted_vllm", + "/v1/responses": "anthropic", +} + + +def test_concurrent_burst_ends_each_stream_by_whether_its_provider_finished_it(gateway: Gateway) -> None: + endings: Final[tuple[Ending, ...]] = ("ending-complete", "ending-clean-eof") + cases: Final = tuple(itertools.product(_BRIDGED_FAMILY, endings, range(5))) + with ExitStack() as wires, gateway.scenario() as scenario: + models: Final = { + route: _deploy(scenario, family, wires.enter_context(wire_server(_upstream(family)))) + for route, family in _BRIDGED_FAMILY.items() + } + with ThreadPoolExecutor(max_workers=12) as pool: + outcomes: Final = tuple( + pool.map(lambda case: _stream_outcome(gateway, models[case[0]], case[0], case[1]), cases) + ) + assert tuple(finished for finished, _ in outcomes) == tuple( + ending == "ending-complete" for _, ending, _ in cases + ), outcomes + assert tuple(spend_row(call_id)["status"] for _, call_id in outcomes) == tuple( + "success" if ending == "ending-complete" else "failure" for _, ending, _ in cases + ) From 17f757892cb3a3596166b6aa4aa57633c06d72de Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:44:26 -0700 Subject: [PATCH 14/22] test(agentcore): inject the async transport instead of patching the aiohttp flag --- .../agentcore/test_agentcore_transformation.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py b/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py index 8ad70d120ac..b89a56041f9 100644 --- a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py +++ b/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py @@ -1299,7 +1299,9 @@ async def _stream_agentcore(body: str, sync_mode: bool) -> list[tuple[str | None ] return [ (chunk.choices[0].delta.content, chunk.choices[0].finish_reason) - async for chunk in await litellm.acompletion(**kwargs, client=AsyncHTTPHandler()) + async for chunk in await litellm.acompletion( + **kwargs, client=AsyncHTTPHandler(transport=httpx.AsyncHTTPTransport()) + ) ] @@ -1310,7 +1312,8 @@ async def _stream_agentcore(body: str, sync_mode: bool) -> list[tuple[str | None pytest.param( 'data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n' 'data: {"event":{"metadata":{"usage":{"inputTokens":9,"outputTokens":7,"totalTokens":16}}}}\n\n' - 'data: {"message":{"role":"assistant","content":[{"text":"Machine learning lets computers learn from data."}]}}\n\n', + 'data: {"message":{"role":"assistant",' + '"content":[{"text":"Machine learning lets computers learn from data."}]}}\n\n', id="strands_full", ), ], @@ -1318,9 +1321,8 @@ async def _stream_agentcore(body: str, sync_mode: bool) -> list[tuple[str | None @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.respx(assert_all_called=True) async def test_agentcore_stream_that_signals_its_end_completes( - ending: str, sync_mode: bool, respx_mock: respx.Router, monkeypatch: pytest.MonkeyPatch + ending: str, sync_mode: bool, respx_mock: respx.Router ) -> None: - monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) respx_mock.post(_AGENTCORE_RUNTIME_URL).mock( return_value=httpx.Response( 200, content=(_AGENTCORE_DELTAS + ending).encode(), headers={"content-type": "text/event-stream"} @@ -1335,10 +1337,7 @@ async def test_agentcore_stream_that_signals_its_end_completes( @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.respx(assert_all_called=True) -async def test_agentcore_stream_cut_before_it_signals_its_end_raises( - sync_mode: bool, respx_mock: respx.Router, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) +async def test_agentcore_stream_cut_before_it_signals_its_end_raises(sync_mode: bool, respx_mock: respx.Router) -> None: respx_mock.post(_AGENTCORE_RUNTIME_URL).mock( return_value=httpx.Response( 200, content=_AGENTCORE_DELTAS.encode(), headers={"content-type": "text/event-stream"} From 3957630777320c94045343565448ed2c5a613e6f Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:44:27 -0700 Subject: [PATCH 15/22] test(guardrails): end the decision model chat fixture with a finish reason, as a real stream does --- .../streaming/test_decision_model_buffering.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/integration/streaming/test_decision_model_buffering.py b/tests/integration/streaming/test_decision_model_buffering.py index a2594fd2b9b..d6fea067d69 100644 --- a/tests/integration/streaming/test_decision_model_buffering.py +++ b/tests/integration/streaming/test_decision_model_buffering.py @@ -229,11 +229,17 @@ def _frames(identity: str, endpoint: str) -> tuple[str, ...]: 'object': 'chat.completion.chunk', 'created': 1, 'model': 'scripted-chat', - 'choices': [{'index': 0, 'delta': {'content': text}, 'finish_reason': None}], + 'choices': [ + { + 'index': 0, + 'delta': {'content': text}, + 'finish_reason': 'stop' if index == len(pieces) - 1 else None, + } + ], } ) }" - for text in pieces + for index, text in enumerate(pieces) ), "data: [DONE]", ) From bfc668a657d082bdef0bf79b2195fe749ac3c12b Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:02:45 -0700 Subject: [PATCH 16/22] test(streaming): shape the AgentCore and LangGraph fixtures like captures from a real runtime and server --- ..._stream_clean_eof_without_finish_reason.py | 26 ++++++++++++++----- .../test_agentcore_transformation.py | 11 +++++--- .../chat/test_langgraph_transformation.py | 10 ++++--- 3 files changed, 33 insertions(+), 14 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index 024783aabfb..03c842ddc99 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -107,19 +107,31 @@ def _gemini_frames(ending: Ending) -> tuple[bytes, ...]: return (*content, usage) if finished else content +def _event(name: str, frame: JsonValue) -> bytes: + return f"event: {name}\n".encode() + _data(frame) + + def _langgraph_frames() -> tuple[bytes, ...]: - """LangGraph ends a run by closing the stream after its metadata event.""" + """A LangGraph server's run stream in its default mode, its run metadata first and then the state after each step""" + question: Final = {"type": "human", "content": "Answer in two steps."} + answers: Final = tuple({"type": "ai", "content": part} for part in _PARTIAL) return ( - *(_data(["messages", [[{"type": "ai", "content": part}, {}]]]) for part in _PARTIAL), - _data(["metadata", {"run_id": "run-1"}]), + _event("metadata", {"run_id": "run-1", "attempt": 1}), + *(_event("values", {"messages": [question, *answers[:step]]}) for step in range(len(answers) + 1)), ) def _agentcore_frames(ending: Ending) -> tuple[bytes, ...]: - """A runtime that is not Strands can end its turn on messageStop with no metadata or final message.""" - content: Final = tuple(_data({"event": {"contentBlockDelta": {"delta": {"text": part}}}}) for part in _PARTIAL) - message_stop: Final = _data({"event": {"messageStop": {"stopReason": "end_turn"}}}) - return (*content, message_stop) if ending == "ending-complete" else content + """A runtime that forwards Bedrock Converse events without the metadata event, so its turn ends on messageStop""" + start: Final = _data({"event": {"messageStart": {"role": "assistant"}}}) + content: Final = tuple( + _data({"event": {"contentBlockDelta": {"delta": {"text": part}, "contentBlockIndex": 0}}}) for part in _PARTIAL + ) + end: Final = ( + _data({"event": {"contentBlockStop": {"contentBlockIndex": 0}}}), + _data({"event": {"messageStop": {"stopReason": "end_turn"}}}), + ) + return (start, *content, *end) if ending == "ending-complete" else (start, *content) def _frames(family: Family, identity: str, ending: Ending) -> tuple[bytes, ...]: diff --git a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py b/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py index b89a56041f9..1b0c789f55d 100644 --- a/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py +++ b/tests/unit/llms/bedrock/chat/agentcore/test_agentcore_transformation.py @@ -1277,8 +1277,9 @@ _AGENTCORE_RUNTIME_URL: Final = ( "arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A888602223428%3Aruntime%2Fhosted_agent_13sf6-cALnp38iZD/invocations" ) _AGENTCORE_DELTAS: Final = ( - 'data: {"event":{"contentBlockDelta":{"delta":{"text":"Machine learning lets "}}}}\n\n' - 'data: {"event":{"contentBlockDelta":{"delta":{"text":"computers learn from data."}}}}\n\n' + 'data: {"event":{"messageStart":{"role":"assistant"}}}\n\n' + 'data: {"event":{"contentBlockDelta":{"delta":{"text":"Machine learning lets "},"contentBlockIndex":0}}}\n\n' + 'data: {"event":{"contentBlockDelta":{"delta":{"text":"computers learn from data."},"contentBlockIndex":0}}}\n\n' ) @@ -1308,7 +1309,11 @@ async def _stream_agentcore(body: str, sync_mode: bool) -> list[tuple[str | None @pytest.mark.parametrize( "ending", [ - pytest.param('data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n', id="message_stop"), + pytest.param( + 'data: {"event":{"contentBlockStop":{"contentBlockIndex":0}}}\n\n' + 'data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n', + id="message_stop", + ), pytest.param( 'data: {"event":{"messageStop":{"stopReason":"end_turn"}}}\n\n' 'data: {"event":{"metadata":{"usage":{"inputTokens":9,"outputTokens":7,"totalTokens":16}}}}\n\n' diff --git a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py index 4db85bc8f58..806ade53080 100644 --- a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py +++ b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py @@ -271,11 +271,13 @@ async def test_langgraph_streaming_acompletion_without_a_client_posts_through_th @pytest.mark.asyncio -async def test_langgraph_async_stream_delivers_every_message_event_before_its_stop(): +async def test_langgraph_async_stream_delivers_every_step_of_the_run_before_its_stop(): async def events(): - yield b'data: ["messages", [[{"type": "ai", "content": "25 * 4"}, {}]]]\n\n' - yield b'data: ["messages", [[{"type": "ai", "content": " is 100"}, {}]]]\n\n' - yield b'data: ["metadata", {"run_id": "run-1"}]\n\n' + question = {"type": "human", "content": "What is 25 * 4?"} + answers = [{"type": "ai", "content": "25 * 4"}, {"type": "ai", "content": " is 100"}] + yield b'event: metadata\ndata: {"run_id": "run-1", "attempt": 1}\n\n' + for step in range(len(answers) + 1): + yield b"event: values\ndata: " + json.dumps({"messages": [question, *answers[:step]]}).encode() + b"\n\n" client = AsyncHTTPHandler(transport=httpx.MockTransport(lambda request: httpx.Response(200, content=events()))) From ada6ff755e95978cd7297cc5aea6db053c006817 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 17:14:18 -0700 Subject: [PATCH 17/22] test(streaming): keep each stream's ending and land every spend row once through a spend log outage --- ..._stream_clean_eof_without_finish_reason.py | 62 ++++++++++++++++++- 1 file changed, 61 insertions(+), 1 deletion(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index 03c842ddc99..1af5e3a4106 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -2,17 +2,22 @@ import asyncio import itertools import json import uuid +from collections import Counter from collections.abc import Callable, Iterator, Mapping from concurrent.futures import ThreadPoolExecutor from contextlib import ExitStack +from pathlib import Path from typing import Final, Literal import anthropic import openai import pytest from integration._support.anthropic_sse import CONTENT_BLOCK_START, CONTENT_TAIL, message_start, text_delta -from integration._support.client import Gateway, Scenario, string_value +from integration._support.client import Gateway, Scenario, eventually, string_value +from integration._support.database import read_rows, scratch_database +from integration._support.database_relay import DatabaseRelay, database_relay from integration._support.openai_wire import answering_model_discovery, posted_targets +from integration._support.process import owned_proxy from integration._support.responses_stream import data_frames, spend_row, sse_events from integration._support.wire import Reply, Request, Wire, wire_server from integration.providers._count_tokens_system_lift import ( @@ -456,3 +461,58 @@ def test_concurrent_burst_ends_each_stream_by_whether_its_provider_finished_it(g assert tuple(spend_row(call_id)["status"] for _, call_id in outcomes) == tuple( "success" if ending == "ending-complete" else "failure" for _, ending, _ in cases ) + + +_SPEND_LOG_INSERT: Final = b'INSERT INTO "public"."LiteLLM_SpendLogs"' +Outcomes = tuple[tuple[bool, str], ...] + + +def _spend_statuses(database_url: str) -> Counter[tuple[str, str]]: + rows: Final = read_rows('SELECT litellm_call_id, status FROM "LiteLLM_SpendLogs"', (), database_url=database_url) + return Counter((string_value(row["litellm_call_id"]), string_value(row["status"])) for row in rows) + + +def _bursts_until_recovered(relay: DatabaseRelay, burst: Callable[[], Outcomes]) -> Iterator[tuple[bool, Outcomes]]: + """Bursts until the proxy reconnects after the outage, each marked by whether the database was down when it began""" + while not relay.reconnected.is_set(): + yield relay.tripped.is_set(), burst() + + +def test_spend_log_outage_mid_burst_keeps_each_stream_ending_and_lands_every_row_once( + gateway: Gateway, tmp_path: Path +) -> None: + endings: Final[tuple[Ending, ...]] = ("ending-complete", "ending-clean-eof") + cases: Final = tuple(itertools.product(_BRIDGED_FAMILY, endings)) + with ( + scratch_database() as database_url, + database_relay(database_url, _SPEND_LOG_INSERT) as (relay, relayed_url), + owned_proxy(gateway, tmp_path, {"DATABASE_URL": relayed_url}, workers=2) as candidate, + ExitStack() as wires, + candidate.scenario() as scenario, + ThreadPoolExecutor(max_workers=len(cases)) as pool, + ): + models: Final = { + route: _deploy(scenario, family, wires.enter_context(wire_server(_upstream(family)))) + for route, family in _BRIDGED_FAMILY.items() + } + + def burst() -> Outcomes: + return tuple(pool.map(lambda case: _stream_outcome(candidate, models[case[0]], case[0], case[1]), cases)) + + relay.arm() + waves: Final = tuple(itertools.islice(_bursts_until_recovered(relay, burst), 60)) + assert relay.reconnected.is_set(), "the proxy never reconnected to the database after the outage" + assert relay.refused >= 1 and any(down for down, _ in waves), (relay.refused, len(waves)) + finished: Final = tuple(ending == "ending-complete" for _, ending in cases) + assert all(tuple(done for done, _ in outcomes) == finished for _, outcomes in waves), waves + sent: Final = tuple(itertools.chain.from_iterable(outcomes for _, outcomes in waves)) + expected: Final = Counter( + (call_id, "success" if ending == "ending-complete" else "failure") + for (_, ending), (_, call_id) in zip(cases * len(waves), sent, strict=True) + ) + landed: Final = eventually( + lambda: _spend_statuses(database_url), + lambda rows: {call_id for call_id, _ in expected} <= {call_id for call_id, _ in rows}, + seconds=120, + ) + assert landed == expected From 1bfb8764fa1786279b01ff8ddd7891be5c166154 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:47:01 -0700 Subject: [PATCH 18/22] test(streaming): end each stream by its provider's finish after a proxy worker is killed mid burst --- ..._stream_clean_eof_without_finish_reason.py | 55 ++++++++++++++++++- 1 file changed, 53 insertions(+), 2 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index 1af5e3a4106..606d1feaaa3 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -10,6 +10,7 @@ from pathlib import Path from typing import Final, Literal import anthropic +import httpx import openai import pytest from integration._support.anthropic_sse import CONTENT_BLOCK_START, CONTENT_TAIL, message_start, text_delta @@ -17,7 +18,7 @@ from integration._support.client import Gateway, Scenario, eventually, string_va from integration._support.database import read_rows, scratch_database from integration._support.database_relay import DatabaseRelay, database_relay from integration._support.openai_wire import answering_model_discovery, posted_targets -from integration._support.process import owned_proxy +from integration._support.process import owned_proxy, owned_proxy_process from integration._support.responses_stream import data_frames, spend_row, sse_events from integration._support.wire import Reply, Request, Wire, wire_server from integration.providers._count_tokens_system_lift import ( @@ -467,7 +468,7 @@ _SPEND_LOG_INSERT: Final = b'INSERT INTO "public"."LiteLLM_SpendLogs"' Outcomes = tuple[tuple[bool, str], ...] -def _spend_statuses(database_url: str) -> Counter[tuple[str, str]]: +def _spend_statuses(database_url: str | None) -> Counter[tuple[str, str]]: rows: Final = read_rows('SELECT litellm_call_id, status FROM "LiteLLM_SpendLogs"', (), database_url=database_url) return Counter((string_value(row["litellm_call_id"]), string_value(row["status"])) for row in rows) @@ -516,3 +517,53 @@ def test_spend_log_outage_mid_burst_keeps_each_stream_ending_and_lands_every_row seconds=120, ) assert landed == expected + + +def _attempt(gateway: Gateway, model: str, route: Route, ending: Ending) -> tuple[bool, str] | None: + """One stream's outcome, or None when the worker serving it was killed before the proxy finished the reply""" + try: + return _stream_outcome(gateway, model, route, ending) + except httpx.TransportError: + return None + + +def test_worker_killed_mid_burst_leaves_the_proxy_ending_each_stream_by_whether_its_provider_finished_it( + gateway: Gateway, tmp_path: Path +) -> None: + endings: Final[tuple[Ending, ...]] = ("ending-complete", "ending-clean-eof") + cases: Final = tuple(itertools.product(_BRIDGED_FAMILY, endings, range(4))) + probes: Final = tuple(itertools.product(_BRIDGED_FAMILY, endings)) + with ( + owned_proxy_process(gateway, tmp_path, {}, workers=2) as owned, + ExitStack() as wires, + owned.gateway.scenario() as scenario, + ): + candidate: Final = owned.gateway + models: Final = { + route: _deploy(scenario, family, wires.enter_context(wire_server(_upstream(family)))) + for route, family in _BRIDGED_FAMILY.items() + } + workers: Final = eventually(owned.worker_pids, lambda pids: len(pids) == 2, seconds=60) + with owned.replacing_worker(min(workers)) as kill, ThreadPoolExecutor(max_workers=len(cases)) as pool: + futures: Final = [ + pool.submit(_attempt, candidate, models[route], route, ending) for route, ending, _ in cases + ] + kill() + attempts: Final = tuple(future.result() for future in futures) + answered: Final = tuple( + (outcome[0], ending == "ending-complete") + for (_, ending, _), outcome in zip(cases, attempts, strict=True) + if outcome is not None + ) + assert all(finished == expected for finished, expected in answered), attempts + after: Final = tuple(_stream_outcome(candidate, models[route], route, ending) for route, ending in probes) + assert tuple(finished for finished, _ in after) == tuple(ending == "ending-complete" for _, ending in probes) + statuses: Final = eventually( + lambda: _spend_statuses(candidate.database_url), + lambda rows: {call_id for _, call_id in after} <= {call_id for call_id, _ in rows}, + seconds=120, + ) + assert tuple( + statuses[(call_id, "success" if ending == "ending-complete" else "failure")] + for (_, ending), (_, call_id) in zip(probes, after, strict=True) + ) == (1,) * len(probes) From e45309a20e71675f7f8539c766d11d5afa6541ed Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:44:25 -0700 Subject: [PATCH 19/22] test(streaming): run the outage test as one burst before and one during the outage instead of a capped wave loop --- ..._stream_clean_eof_without_finish_reason.py | 25 ++++++++----------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index 606d1feaaa3..ee3c74292b4 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -16,7 +16,7 @@ import pytest from integration._support.anthropic_sse import CONTENT_BLOCK_START, CONTENT_TAIL, message_start, text_delta from integration._support.client import Gateway, Scenario, eventually, string_value from integration._support.database import read_rows, scratch_database -from integration._support.database_relay import DatabaseRelay, database_relay +from integration._support.database_relay import database_relay from integration._support.openai_wire import answering_model_discovery, posted_targets from integration._support.process import owned_proxy, owned_proxy_process from integration._support.responses_stream import data_frames, spend_row, sse_events @@ -465,7 +465,6 @@ def test_concurrent_burst_ends_each_stream_by_whether_its_provider_finished_it(g _SPEND_LOG_INSERT: Final = b'INSERT INTO "public"."LiteLLM_SpendLogs"' -Outcomes = tuple[tuple[bool, str], ...] def _spend_statuses(database_url: str | None) -> Counter[tuple[str, str]]: @@ -473,12 +472,6 @@ def _spend_statuses(database_url: str | None) -> Counter[tuple[str, str]]: return Counter((string_value(row["litellm_call_id"]), string_value(row["status"])) for row in rows) -def _bursts_until_recovered(relay: DatabaseRelay, burst: Callable[[], Outcomes]) -> Iterator[tuple[bool, Outcomes]]: - """Bursts until the proxy reconnects after the outage, each marked by whether the database was down when it began""" - while not relay.reconnected.is_set(): - yield relay.tripped.is_set(), burst() - - def test_spend_log_outage_mid_burst_keeps_each_stream_ending_and_lands_every_row_once( gateway: Gateway, tmp_path: Path ) -> None: @@ -497,19 +490,21 @@ def test_spend_log_outage_mid_burst_keeps_each_stream_ending_and_lands_every_row for route, family in _BRIDGED_FAMILY.items() } - def burst() -> Outcomes: + def burst() -> tuple[tuple[bool, str], ...]: return tuple(pool.map(lambda case: _stream_outcome(candidate, models[case[0]], case[0], case[1]), cases)) relay.arm() - waves: Final = tuple(itertools.islice(_bursts_until_recovered(relay, burst), 60)) - assert relay.reconnected.is_set(), "the proxy never reconnected to the database after the outage" - assert relay.refused >= 1 and any(down for down, _ in waves), (relay.refused, len(waves)) + before: Final = burst() + assert relay.tripped.wait(90), "no spend log write reached the database after the first burst" + down: Final = not relay.reconnected.is_set() + during: Final = burst() + assert relay.reconnected.wait(90), "the proxy never reconnected to the database after the outage" + assert down and relay.refused >= 1, relay.refused finished: Final = tuple(ending == "ending-complete" for _, ending in cases) - assert all(tuple(done for done, _ in outcomes) == finished for _, outcomes in waves), waves - sent: Final = tuple(itertools.chain.from_iterable(outcomes for _, outcomes in waves)) + assert tuple(done for done, _ in before + during) == finished * 2, (before, during) expected: Final = Counter( (call_id, "success" if ending == "ending-complete" else "failure") - for (_, ending), (_, call_id) in zip(cases * len(waves), sent, strict=True) + for (_, ending), (_, call_id) in zip(cases * 2, before + during, strict=True) ) landed: Final = eventually( lambda: _spend_statuses(database_url), From 498aaba60feaf345ba9bac105300a3e9fb3d54d2 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Wed, 21 Oct 2026 03:03:17 -0700 Subject: [PATCH 20/22] test(streaming): connect the outage test's proxy to the relay without TLS so its trigger sees the spend-log insert on an ssl=on Postgres --- .../streaming/test_stream_clean_eof_without_finish_reason.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py index ee3c74292b4..f3b427d91c4 100644 --- a/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py +++ b/tests/integration/streaming/test_stream_clean_eof_without_finish_reason.py @@ -16,7 +16,7 @@ import pytest from integration._support.anthropic_sse import CONTENT_BLOCK_START, CONTENT_TAIL, message_start, text_delta from integration._support.client import Gateway, Scenario, eventually, string_value from integration._support.database import read_rows, scratch_database -from integration._support.database_relay import database_relay +from integration._support.database_relay import database_relay, plaintext_database_url from integration._support.openai_wire import answering_model_discovery, posted_targets from integration._support.process import owned_proxy, owned_proxy_process from integration._support.responses_stream import data_frames, spend_row, sse_events @@ -480,7 +480,7 @@ def test_spend_log_outage_mid_burst_keeps_each_stream_ending_and_lands_every_row with ( scratch_database() as database_url, database_relay(database_url, _SPEND_LOG_INSERT) as (relay, relayed_url), - owned_proxy(gateway, tmp_path, {"DATABASE_URL": relayed_url}, workers=2) as candidate, + owned_proxy(gateway, tmp_path, {"DATABASE_URL": plaintext_database_url(relayed_url)}, workers=2) as candidate, ExitStack() as wires, candidate.scenario() as scenario, ThreadPoolExecutor(max_workers=len(cases)) as pool, From f806780ca7b4e8ad1de489dcf488b2a5604b5a85 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Tue, 27 Oct 2026 01:58:40 -0700 Subject: [PATCH 21/22] test(streaming): type the new stream EOF unit tests --- .../test_streaming_handler.py | 24 ++++++++++--------- .../chat/test_langgraph_transformation.py | 18 ++++++++------ 2 files changed, 24 insertions(+), 18 deletions(-) diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 12b14fd41b1..da115694af0 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -1682,11 +1682,11 @@ def _incomplete_openai_chat_chunk() -> ModelResponseStream: @pytest.mark.asyncio async def test_async_openai_compatible_chat_clean_eof_without_finish_reason_raises_midstream_fallback( logging_obj: Logging, custom_llm_provider: str -): +) -> None: """An OpenAI-compatible stream without a provider finish reason must fail at clean EOF.""" logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider - response = CustomStreamWrapper( + response: Final = CustomStreamWrapper( completion_stream=_as_async(iter([_incomplete_openai_chat_chunk()])), model="gpt-5.6", logging_obj=logging_obj, @@ -1694,7 +1694,7 @@ async def test_async_openai_compatible_chat_clean_eof_without_finish_reason_rais stream_options={"include_usage": True}, ) - first_chunk = await response.__anext__() + first_chunk: Final = await response.__anext__() with pytest.raises(MidStreamFallbackError) as excinfo: await response.__anext__() @@ -1707,11 +1707,11 @@ async def test_async_openai_compatible_chat_clean_eof_without_finish_reason_rais @pytest.mark.parametrize("custom_llm_provider", ["openai", "azure", "hosted_vllm"]) def test_sync_openai_compatible_chat_clean_eof_without_finish_reason_raises_midstream_fallback( logging_obj: Logging, custom_llm_provider: str -): +) -> None: """The synchronous OpenAI-compatible stream must fail at clean EOF too.""" logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider - response = CustomStreamWrapper( + response: Final = CustomStreamWrapper( completion_stream=iter([_incomplete_openai_chat_chunk()]), model="gpt-5.6", logging_obj=logging_obj, @@ -1719,7 +1719,7 @@ def test_sync_openai_compatible_chat_clean_eof_without_finish_reason_raises_mids stream_options={"include_usage": True}, ) - first_chunk = next(response) + first_chunk: Final = next(response) with pytest.raises(MidStreamFallbackError) as excinfo: next(response) @@ -1740,7 +1740,7 @@ async def _collect(wrapper: CustomStreamWrapper, sync_mode: bool) -> list[ModelR @pytest.mark.asyncio async def test_stream_that_sends_no_chunks_keeps_the_default_stop( logging_obj: Logging, custom_llm_provider: str, sync_mode: bool -): +) -> None: """Gemini's OpenAI-compatible endpoint answers with only [DONE] when reasoning uses up max_tokens.""" wrapper: Final = CustomStreamWrapper( @@ -1806,7 +1806,7 @@ def _anthropic_stream( @pytest.mark.asyncio async def test_anthropic_stream_that_ends_before_message_delta_fails_with_the_eof_message( sync_mode: bool, logging_obj: Logging -): +) -> None: wrapper: Final = _anthropic_stream(_ANTHROPIC_PARTIAL_EVENTS, sync_mode, logging_obj) with pytest.raises(MidStreamFallbackError) as excinfo: @@ -1821,7 +1821,9 @@ async def test_anthropic_stream_that_ends_before_message_delta_fails_with_the_eo @pytest.mark.parametrize("sync_mode", [True, False]) @pytest.mark.asyncio -async def test_anthropic_stream_with_message_stop_ends_with_its_finish_reason(sync_mode: bool, logging_obj: Logging): +async def test_anthropic_stream_with_message_stop_ends_with_its_finish_reason( + sync_mode: bool, logging_obj: Logging +) -> None: wrapper: Final = _anthropic_stream( (*_ANTHROPIC_PARTIAL_EVENTS, *_ANTHROPIC_TERMINAL_EVENTS), sync_mode, logging_obj ) @@ -1838,7 +1840,7 @@ async def test_anthropic_stream_with_message_stop_ends_with_its_finish_reason(sy @pytest.mark.asyncio async def test_generic_chunk_stream_without_a_finish_reason_keeps_the_default_stop( sync_mode: bool, logging_obj: Logging -): +) -> None: """Providers that stream GenericStreamingChunks and never mark the last one finished still end with stop, but the logged response built from ``chunks`` never gains that synthetic stop.""" chunks: Final = [ @@ -1863,7 +1865,7 @@ async def test_generic_chunk_stream_without_a_finish_reason_keeps_the_default_st @pytest.mark.asyncio async def test_gemini_stream_with_a_usage_chunk_after_its_finish_reason_still_completes( sync_mode: bool, logging_obj: Logging -): +) -> None: """Gemini marks every chunk unfinished, so its finishReason only shows up mid-stream before the usage chunk.""" lines: Final = [ f"data: {json.dumps(event)}" diff --git a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py index 806ade53080..256876b7eea 100644 --- a/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py +++ b/tests/unit/llms/langgraph/chat/test_langgraph_transformation.py @@ -22,6 +22,8 @@ import asyncio import importlib import json import os +from collections.abc import AsyncIterator +from typing import Final import httpx import pytest @@ -271,23 +273,25 @@ async def test_langgraph_streaming_acompletion_without_a_client_posts_through_th @pytest.mark.asyncio -async def test_langgraph_async_stream_delivers_every_step_of_the_run_before_its_stop(): - async def events(): - question = {"type": "human", "content": "What is 25 * 4?"} - answers = [{"type": "ai", "content": "25 * 4"}, {"type": "ai", "content": " is 100"}] +async def test_langgraph_async_stream_delivers_every_step_of_the_run_before_its_stop() -> None: + async def events() -> AsyncIterator[bytes]: + question: Final = {"type": "human", "content": "What is 25 * 4?"} + answers: Final = ({"type": "ai", "content": "25 * 4"}, {"type": "ai", "content": " is 100"}) yield b'event: metadata\ndata: {"run_id": "run-1", "attempt": 1}\n\n' for step in range(len(answers) + 1): yield b"event: values\ndata: " + json.dumps({"messages": [question, *answers[:step]]}).encode() + b"\n\n" - client = AsyncHTTPHandler(transport=httpx.MockTransport(lambda request: httpx.Response(200, content=events()))) + client: Final = AsyncHTTPHandler( + transport=httpx.MockTransport(lambda request: httpx.Response(200, content=events())) + ) - response = await litellm.acompletion( + response: Final = await litellm.acompletion( model="langgraph/agent", messages=[{"role": "user", "content": "What is 25 * 4?"}], api_base="http://langgraph.test", stream=True, client=client, ) - streamed = [(chunk.choices[0].delta.content, chunk.choices[0].finish_reason) async for chunk in response] + streamed: Final = [(chunk.choices[0].delta.content, chunk.choices[0].finish_reason) async for chunk in response] assert streamed == [("25 * 4", None), (" is 100", None), (None, "stop")] From 9dae6bc70bcd241482a83cbc8c2271463dbbca25 Mon Sep 17 00:00:00 2001 From: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com> Date: Fri, 30 Oct 2026 12:56:00 -0700 Subject: [PATCH 22/22] test(streaming): a merged reasoning stream cut mid reasoning now fails Main's think tag test expected a stream that ends mid reasoning without a finish reason to finish with a closed think block. With this PR that stream fails, so the case moves to its own test that expects MidStreamFallbackError with the open think block as the partial content --- .../test_streaming_handler.py | 26 ++++++++++++++----- 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index ca5f9c5a63f..e3201263a3c 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -576,11 +576,6 @@ _GET_WEATHER_CALL: Final = { "abcd", 0, ), - ( - (_stream_chunk({"reasoning_content": "Still thinking"}),), - "Still thinking", - 0, - ), ( (_stream_chunk({"content": "hi"}), _stream_chunk({}, "stop")), "hi", @@ -592,7 +587,6 @@ _GET_WEATHER_CALL: Final = { "reasoning_then_function_call", "reasoning_cut_by_length", "reasoning_after_answer", - "stream_ends_mid_reasoning", "answer_without_reasoning", ], ) @@ -626,6 +620,26 @@ async def test_merged_reasoning_stream_keeps_think_tags_balanced( ) +@pytest.mark.asyncio +@pytest.mark.parametrize("sync_mode", [True, False]) +async def test_merged_reasoning_stream_cut_mid_reasoning_fails_instead_of_closing_the_think_block( + logging_obj: Logging, sync_mode: bool +) -> None: + response: Final = CustomStreamWrapper( + completion_stream=ModelResponseListIterator( + model_responses=[_stream_chunk({"reasoning_content": "Still thinking"})] + ), + model="deepseek-reasoner", + logging_obj=logging_obj, + ) + response.merge_reasoning_content_in_choices = True + + with pytest.raises(MidStreamFallbackError) as excinfo: + await _collect(response, sync_mode) + + assert excinfo.value.generated_content == "Still thinking" + + def test_strip_sse_data_from_chunk(): """Test the static method that strips 'data: ' prefix from SSE chunks""" # Test with string inputs