From 5cbbade24c87fc9310bab43986b385a670b48113 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Tue, 14 Jul 2026 23:55:31 +0530 Subject: [PATCH] fix(streaming): zero-event guard parity for the anthropic_messages path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The chat_completions path raises EmptyStreamError when a stream yields no chunks and no finish_reason (#64420). _call_anthropic() had no equivalent, and the eventless failure surfaces differently by client: - Real Anthropic SDK: no message_start means no final-message snapshot, so get_final_message() raises a bare AssertionError — not in the retry loop's transient set, so it burned no retries and surfaced raw. - OpenAI-compat shims: may fabricate a contentless Message with no stop_reason (or return None), flowing out as a 'successful' empty turn. Track whether any stream event arrived and normalize both shapes to EmptyStreamError, giving the anthropic path the same transient retry budget in the shared _call() retry loop. A real completed response always carries a stop_reason, so the guard cannot fire on legitimate turns (including eventless mocks with stop_reason='end_turn'). Follow-up to #64420. --- agent/chat_completion_helpers.py | 45 ++++++++++++++++- tests/run_agent/test_streaming.py | 84 +++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 1 deletion(-) diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 9f114d7e6528..dbc946ce899a 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -2623,6 +2623,17 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= works unchanged. """ has_tool_use = False + # Zero-event guard parity with the chat_completions path: track + # whether the provider delivered ANY stream event. On an eventless + # stream the real Anthropic SDK's get_final_message() raises + # AssertionError (no message_start ⇒ no final-message snapshot); + # OpenAI-compat shims may instead fabricate a contentless Message + # with no stop_reason, or return None under ``python -O`` (assert + # stripped). Every one of those shapes is normalized below to + # EmptyStreamError so the shared _call() retry loop treats it as + # transient instead of surfacing a raw AssertionError or a + # fabricated "successful" empty turn. + saw_stream_event = False # Reset stale-stream timer for this attempt last_chunk_time["t"] = time.time() @@ -2650,6 +2661,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= except Exception: pass for event in stream: + saw_stream_event = True # Update stale-stream timer on every event so the # outer poll loop knows data is flowing. Without # this, the detector kills healthy long-running @@ -2710,7 +2722,38 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= # this return value is discarded anyway. if agent._interrupt_requested: return None - return stream.get_final_message() + # Zero-event guard (parity with the chat_completions zero-chunk + # guard above). Real SDK: an eventless stream has no + # message_start, so get_final_message() raises AssertionError + # (final-message snapshot is None) — normalize that to + # EmptyStreamError so it gets the transient retry budget + # instead of surfacing raw. + try: + _final_message = stream.get_final_message() + except AssertionError: + if not saw_stream_event: + raise EmptyStreamError( + "Provider returned an empty stream with no events " + "(possible upstream error or malformed event stream)." + ) from None + raise + # Shim variants of the same failure: an OpenAI-compat adapter + # may fabricate a contentless Message with no stop_reason, or + # return None where the SDK assert would have fired (e.g. + # ``python -O``). A real completed response always carries a + # stop_reason, so this cannot fire on legitimate turns. + if not saw_stream_event and ( + _final_message is None + or ( + not getattr(_final_message, "content", None) + and getattr(_final_message, "stop_reason", None) is None + ) + ): + raise EmptyStreamError( + "Provider returned an empty stream with no stop_reason " + "(possible upstream error or malformed event stream)." + ) + return _final_message def _call(): import httpx as _httpx diff --git a/tests/run_agent/test_streaming.py b/tests/run_agent/test_streaming.py index f91a5aaa844d..cdbcd51828bd 100644 --- a/tests/run_agent/test_streaming.py +++ b/tests/run_agent/test_streaming.py @@ -1293,6 +1293,90 @@ class TestAnthropicStreamCallbacks: assert agent._anthropic_client.messages.stream.call_count == 1 assert mock_replace.call_count == 0 + @patch("run_agent.AIAgent._try_refresh_anthropic_client_credentials") + @patch("run_agent.AIAgent._rebuild_anthropic_client") + @patch("run_agent.AIAgent._replace_primary_openai_client") + def test_anthropic_zero_event_stream_retried_as_transient( + self, mock_replace, mock_rebuild, mock_refresh, + ): + """An eventless Anthropic stream with an empty final Message gets the + same transient retry budget as the chat_completions zero-chunk guard + (parity follow-up to #64420).""" + from agent.errors import EmptyStreamError + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="https://api.anthropic.com", + provider="anthropic", + model="claude-test", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent.api_mode = "anthropic_messages" + agent._interrupt_requested = False + + empty_message = SimpleNamespace(content=[], stop_reason=None) + empty_stream = MagicMock() + empty_stream.__enter__ = MagicMock(return_value=empty_stream) + empty_stream.__exit__ = MagicMock(return_value=False) + empty_stream.__iter__ = MagicMock(side_effect=lambda: iter([])) + empty_stream.get_final_message.return_value = empty_message + + agent._anthropic_client = MagicMock() + agent._anthropic_client.messages.stream.return_value = empty_stream + + with pytest.raises(EmptyStreamError): + agent._interruptible_streaming_api_call({}) + + assert agent._anthropic_client.messages.stream.call_count == 3 + # Anthropic-native cleanup between attempts: rebuild the Anthropic + # client, never the OpenAI primary client. + assert mock_replace.call_count == 0 + assert mock_rebuild.call_count == 2 + + @patch("run_agent.AIAgent._try_refresh_anthropic_client_credentials") + @patch("run_agent.AIAgent._rebuild_anthropic_client") + @patch("run_agent.AIAgent._replace_primary_openai_client") + def test_anthropic_eventless_sdk_assertion_normalized_to_empty_stream( + self, mock_replace, mock_rebuild, mock_refresh, + ): + """Real-SDK shape: an eventless stream has no message_start, so + get_final_message() raises AssertionError (final snapshot is None). + That must be normalized to EmptyStreamError and retried as + transient — not surface as a raw AssertionError.""" + from agent.errors import EmptyStreamError + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="https://api.anthropic.com", + provider="anthropic", + model="claude-test", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent.api_mode = "anthropic_messages" + agent._interrupt_requested = False + + empty_stream = MagicMock() + empty_stream.__enter__ = MagicMock(return_value=empty_stream) + empty_stream.__exit__ = MagicMock(return_value=False) + empty_stream.__iter__ = MagicMock(side_effect=lambda: iter([])) + empty_stream.get_final_message.side_effect = AssertionError() + + agent._anthropic_client = MagicMock() + agent._anthropic_client.messages.stream.return_value = empty_stream + + with pytest.raises(EmptyStreamError): + agent._interruptible_streaming_api_call({}) + + assert agent._anthropic_client.messages.stream.call_count == 3 + assert mock_replace.call_count == 0 + assert mock_rebuild.call_count == 2 + class TestPartialToolCallWarning: """Regression: when a stream dies mid tool-call argument generation after