[Bugfix] Mark Omni chat benchmark streams with error events as failed (#8000)

Signed-off-by: levius <2114377220@qq.com>

tests/benchmarks/patch/test_patch.py

@@ -434,6 +434,45 @@ async def test_bundled_first_text_chunk_uses_stage0_token_timings(mocker: Mocker
     assert output.tpot_measured is True
 
 
+@pytest.mark.asyncio
+async def test_streaming_error_chunk_marks_request_failed(mocker: MockerFixture):
+    """HTTP 200 streams can still terminate with an OpenAI error event."""
+    request_input = RequestFuncInput(
+        model="test-model",
+        model_name="test-model",
+        prompt="test prompt",
+        api_url="http://test.com/v1/chat/completions",
+        prompt_len=10,
+        output_len=20,
+    )
+    chunks = [
+        create_sse_chunk(
+            {
+                "choices": [{"delta": {"content": "partial response"}}],
+                "modality": "text",
+            }
+        ),
+        create_sse_chunk(
+            {
+                "error": {
+                    "message": "EngineCore encountered an issue",
+                    "type": "BadRequestError",
+                    "code": 400,
+                }
+            }
+        ),
+        b"data: [DONE]\n\n",
+    ]
+    mock_response = MockResponse(200, chunks)
+    mock_session = mocker.AsyncMock()
+    mock_session.post = mocker.MagicMock(return_value=mock_response)
+
+    output = await async_request_openai_chat_omni_completions(request_input, mock_session)
+
+    assert output.success is False
+    assert output.error == "EngineCore encountered an issue"
+
+
 @pytest.mark.asyncio
 async def test_positive_client_text_timings_take_precedence_over_stage0(mocker: MockerFixture):
     request_input = RequestFuncInput(

vllm_omni/benchmarks/patch/patch.py

@@ -1862,6 +1862,7 @@ async def async_request_openai_chat_omni_completions(
         output.peak_memory_mb = 0.0
         output.stage_durations = None
         completion_tokens_seen = 0
+        streaming_error_received = False
         try:
             async with session.post(url=api_url, json=payload, headers=headers) as response:
                 if response.status == 200:
@@ -1889,6 +1890,13 @@ async def async_request_openai_chat_omni_completions(
                             if chunk != "[DONE]":
                                 timestamp = time.perf_counter()
                                 data = json.loads(chunk)
+                                if (streaming_error := data.get("error")) is not None:
+                                    streaming_error_received = True
+                                    if isinstance(streaming_error, dict):
+                                        output.error = str(streaming_error.get("message") or streaming_error)
+                                    else:
+                                        output.error = str(streaming_error)
+                                    continue
                                 _update_output_stage_metrics_from_payload(output, data)
                                 _update_output_peak_memory_from_payload(output, data)
                                 _update_output_stage_durations_from_payload(output, data)
@@ -2073,7 +2081,7 @@ async def async_request_openai_chat_omni_completions(
                                     output.tts_output_pcm_bytes = (waveform * 32767).astype(np.int16).tobytes()
                             except Exception as ex:
                                 logger.warning("seed_tts WER PCM export failed: %s", ex)
-                    output.success = True
+                    output.success = not streaming_error_received
                 else:
                     output.error = response.reason or ""
                     output.success = False