From 0fcebb29f442544e1ecafba618abde6e2f5fa197 Mon Sep 17 00:00:00 2001
From: beiyesi <shiqiming.sqm@taobao.com>
Date: Thu, 2 Jul 2026 12:07:50 +0800
Subject: [PATCH] fix streaming bare data error payloads

---
 agent/chat_completion_helpers.py  |  9 +++++-
 tests/run_agent/test_run_agent.py | 54 +++++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+), 1 deletion(-)

diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py
index 741372bb81..c4104e7769 100644
--- a/agent/chat_completion_helpers.py
+++ b/agent/chat_completion_helpers.py
@@ -330,8 +330,15 @@ def _provider_stream_error_from_text(
         is_structured_error_event = is_error_event and (
             has_error_finish or is_http_error or is_error_payload
         )
+        is_bare_error_finish_payload = (
+            not is_error_event and has_error_finish and is_error_payload
+        )
 
-        if not (is_http_error or is_structured_error_event):
+        if not (
+            is_http_error
+            or is_structured_error_event
+            or is_bare_error_finish_payload
+        ):
             continue
 
         headers = getattr(response, "headers", None) if response is not None else None
diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py
index 6a01c21e12..775d6504f8 100644
--- a/tests/run_agent/test_run_agent.py
+++ b/tests/run_agent/test_run_agent.py
@@ -5569,6 +5569,13 @@ def _provider_sse_error_text(status=503, code="ServiceUnavailable", message="Bus
     )
 
 
+def _provider_bare_sse_error_text(
+    code="rate_limit_exceeded",
+    message="Rate limit exceeded.",
+):
+    return f'data: {{"error":{{"code":"{code}","message":"{message}"}}}}\n'
+
+
 class TestStreamingApiCall:
     """Tests for _streaming_api_call — voice TTS streaming pipeline."""
 
@@ -5628,6 +5635,23 @@ class TestStreamingApiCall:
         assert "Busy" in str(exc)
         agent.stream_delta_callback.assert_not_called()
 
+    def test_error_finish_bare_sse_error_payload_raises_provider_error(self, agent):
+        chunks = [
+            _make_chunk(content=_provider_bare_sse_error_text()),
+            _make_chunk(finish_reason="error_finish"),
+        ]
+        agent.client.chat.completions.create.return_value = iter(chunks)
+        agent.stream_delta_callback = MagicMock()
+
+        with pytest.raises(Exception) as exc_info:
+            agent._interruptible_streaming_api_call({"messages": []})
+
+        exc = exc_info.value
+        assert getattr(exc, "status_code", None) is None
+        assert getattr(exc, "body", {})["error"]["code"] == "rate_limit_exceeded"
+        assert "Rate limit exceeded" in str(exc)
+        agent.stream_delta_callback.assert_not_called()
+
     def test_provider_error_prefix_like_normal_text_flushes_to_callback(self, agent):
         chunks = [
             _make_chunk(content="id: product-42\n"),
@@ -5659,6 +5683,36 @@ class TestStreamingApiCall:
             call.args[0] for call in agent.stream_delta_callback.call_args_list
         ] == ['event: error\ndata: {"example": true}\n']
 
+    def test_bare_sse_error_payload_with_stop_is_literal_text(self, agent):
+        error_text = _provider_bare_sse_error_text(message="Example error payload.")
+        chunks = [
+            _make_chunk(content=error_text),
+            _make_chunk(finish_reason="stop"),
+        ]
+        agent.client.chat.completions.create.return_value = iter(chunks)
+        agent.stream_delta_callback = MagicMock()
+
+        resp = agent._interruptible_streaming_api_call({"messages": []})
+
+        assert resp.choices[0].message.content == error_text
+        assert [
+            call.args[0] for call in agent.stream_delta_callback.call_args_list
+        ] == [error_text]
+
+    def test_bare_sse_error_payload_without_finish_reason_is_literal_text(self, agent):
+        error_text = _provider_bare_sse_error_text(message="Example error payload.")
+        chunks = [_make_chunk(content=error_text)]
+        agent.client.chat.completions.create.return_value = iter(chunks)
+        agent.stream_delta_callback = MagicMock()
+
+        resp = agent._interruptible_streaming_api_call({"messages": []})
+
+        assert resp.choices[0].message.content == error_text
+        assert resp.choices[0].finish_reason == "stop"
+        assert [
+            call.args[0] for call in agent.stream_delta_callback.call_args_list
+        ] == [error_text]
+
     def test_run_conversation_retries_stream_error_finish_rate_limit(self, agent):
         first_attempt = iter([
             _make_chunk(content=_provider_sse_429_text()),