diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 9a4bc8a1fe6c..2204c44f925a 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -28,6 +28,7 @@ from typing import Any, Dict, Optional from hermes_cli.timeouts import get_provider_request_timeout, get_provider_stale_timeout from hermes_constants import PARTIAL_STREAM_STUB_ID, FINISH_REASON_LENGTH from agent.error_classifier import FailoverReason +from agent.errors import EmptyStreamError from agent.gemini_native_adapter import is_native_gemini_base_url from agent.model_metadata import is_local_endpoint from agent.message_sanitization import ( @@ -2533,7 +2534,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= and not reasoning_parts and not tool_calls_acc ): - raise RuntimeError( + raise EmptyStreamError( "Provider returned an empty stream with no finish_reason " "(possible upstream error or malformed SSE response)." ) @@ -2756,6 +2757,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= e, (_httpx.ConnectError, _httpx.RemoteProtocolError, ConnectionError) ) _is_stream_parse_err = agent._is_provider_stream_parse_error(e) + _is_empty_stream = isinstance(e, EmptyStreamError) # If the stream died AFTER some tokens were delivered: # normally we don't retry (the user already saw text, @@ -2898,7 +2900,13 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= for phrase in _SSE_CONN_PHRASES ) - if _is_timeout or _is_conn_err or _is_sse_conn_err or _is_stream_parse_err: + if ( + _is_timeout + or _is_conn_err + or _is_sse_conn_err + or _is_stream_parse_err + or _is_empty_stream + ): # Transient network / timeout error. Retry the # streaming request with a fresh connection first. if _stream_attempt < _max_stream_retries: diff --git a/agent/errors.py b/agent/errors.py index abedd83d29ff..89afcd3b4c8a 100644 --- a/agent/errors.py +++ b/agent/errors.py @@ -1,3 +1,9 @@ class SSLConfigurationError(Exception): """Raised when SSL/TLS certificate bundle configuration fails.""" pass + + +class EmptyStreamError(RuntimeError): + """Raised when a provider closes a stream without yielding a response.""" + + pass diff --git a/tests/run_agent/test_streaming.py b/tests/run_agent/test_streaming.py index 7ccafff665a4..f91a5aaa844d 100644 --- a/tests/run_agent/test_streaming.py +++ b/tests/run_agent/test_streaming.py @@ -771,6 +771,34 @@ class TestStreamingFallback: assert mock_client.chat.completions.create.call_count == 3 assert mock_close.call_count >= 1 + @patch("run_agent.AIAgent._create_request_openai_client") + @patch("run_agent.AIAgent._close_request_openai_client") + def test_zero_chunk_stream_retried_as_transient(self, mock_close, mock_create): + """A stream that yields no chunks gets the same retry budget as a drop.""" + from agent.errors import EmptyStreamError + from run_agent import AIAgent + + mock_client = MagicMock() + mock_client.chat.completions.create.return_value = iter(()) + mock_create.return_value = mock_client + + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + model="test/model", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent.api_mode = "chat_completions" + agent._interrupt_requested = False + + with pytest.raises(EmptyStreamError): + agent._interruptible_streaming_api_call({}) + + assert mock_client.chat.completions.create.call_count == 3 + assert mock_close.call_count >= 1 + @patch("run_agent.AIAgent._create_request_openai_client") @patch("run_agent.AIAgent._close_request_openai_client") def test_sse_connection_lost_retried_as_transient(self, mock_close, mock_create):