fix(streaming): zero-event guard parity for the anthropic_messages path

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

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

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