From 5ec576ab4c241afa8b56bee729776ccf7a0eb5eb Mon Sep 17 00:00:00 2001 From: psv666 <150513104+psv666@users.noreply.github.com> Date: Wed, 26 Aug 2026 18:27:25 +0800 Subject: [PATCH] [Bugfix] Fix missing Realtime audio completion (#6564) Signed-off-by: psv666 <2693925048@qq.com> Co-authored-by: amy-why-3459 --- tests/engine/test_orchestrator.py | 164 ++++++++++++++++++ .../test_qwen3_omni_realtime_websocket.py | 46 ++++- vllm_omni/engine/orchestrator.py | 62 ++++++- 3 files changed, 260 insertions(+), 12 deletions(-) diff --git a/tests/engine/test_orchestrator.py b/tests/engine/test_orchestrator.py index 7103a4948..61a32fb86 100644 --- a/tests/engine/test_orchestrator.py +++ b/tests/engine/test_orchestrator.py @@ -17,6 +17,7 @@ import janus import pytest from vllm.outputs import CompletionOutput, RequestOutput from vllm.sampling_params import SamplingParams +from vllm.v1.engine import EngineCoreOutput, EngineCoreOutputs, FinishReason from vllm.v1.engine.exceptions import EngineDeadError from vllm.v1.metrics.stats import IterationStats @@ -106,6 +107,13 @@ class OrchestratorFixture: result_future: concurrent.futures.Future[None] +@dataclass +class FakePromptRequest: + request_id: str + prompt_token_ids: list[int] + resumable: bool = True + + class FakeStageClient: def __init__( self, @@ -291,6 +299,20 @@ def _engine_core_outputs(tag: str, timestamp: float) -> SimpleNamespace: return SimpleNamespace(outputs=[tag], timestamp=timestamp, scheduler_stats=None, finished_requests=None) +def _terminal_engine_core_outputs(request_id: str) -> EngineCoreOutputs: + return EngineCoreOutputs( + outputs=[ + EngineCoreOutput( + request_id=request_id, + new_token_ids=[], + finish_reason=FinishReason.STOP, + ) + ], + timestamp=1.0, + finished_requests={request_id}, + ) + + def _build_request_output( request_id: str, *, @@ -764,6 +786,148 @@ async def test_run_async_chunk(orchestrator_factory) -> None: await _shutdown_orchestrator(orchestrator_fixture) +@pytest.mark.asyncio +async def test_async_chunk_raw_terminal_without_processed_output_finishes_request(orchestrator_factory) -> None: + stage0 = FakeStageClient(stage_type="llm", final_output=False) + stage1 = FakeStageClient(stage_type="llm", final_output=True, final_output_type="audio") + orchestrator_fixture = orchestrator_factory( + [stage0, stage1], + output_processors=[FakeOutputProcessor(), FakeOutputProcessor()], + async_chunk=True, + ) + request = FakePromptRequest( + request_id="req-stream-terminal", + prompt_token_ids=[1, 2, 3, 4], + ) + + try: + await _enqueue_add_request( + orchestrator_fixture, + request_id=request.request_id, + prompt=request, + original_prompt={"prompt": "stream audio"}, + sampling_params_list=[_sampling_params(), _sampling_params()], + final_stage_id=1, + ) + + await _wait_for(lambda: len(stage1.add_request_calls) == 1) + stage1.push_engine_core_outputs(_terminal_engine_core_outputs(request.request_id)) + + output_msg = await _get_output_message(orchestrator_fixture) + + assert output_msg.request_id == request.request_id + assert output_msg.stage_id == 1 + assert output_msg.finished is True + assert output_msg.engine_outputs.finished is True + assert output_msg.engine_outputs.outputs[0].multimodal_output["audio"].numel() == 0 + await _wait_for(lambda: request.request_id not in orchestrator_fixture.orchestrator.request_states) + finally: + await _shutdown_orchestrator(orchestrator_fixture) + + +@pytest.mark.asyncio +async def test_async_chunk_data_then_raw_terminal_finishes_once(orchestrator_factory) -> None: + request_id = "req-stream-data-terminal" + stage0 = FakeStageClient(stage_type="llm", final_output=False) + stage1 = FakeStageClient(stage_type="llm", final_output=True, final_output_type="audio") + final_data = _build_request_output( + request_id, + token_ids=[20, 21], + finished=False, + text="last audio chunk", + ) + orchestrator_fixture = orchestrator_factory( + [stage0, stage1], + output_processors=[ + FakeOutputProcessor(), + FakeOutputProcessor(request_outputs=[final_data]), + ], + async_chunk=True, + ) + request = FakePromptRequest( + request_id=request_id, + prompt_token_ids=[1, 2, 3, 4], + ) + + try: + await _enqueue_add_request( + orchestrator_fixture, + request_id=request_id, + prompt=request, + original_prompt={"prompt": "stream audio"}, + sampling_params_list=[_sampling_params(), _sampling_params()], + final_stage_id=1, + ) + + await _wait_for(lambda: len(stage1.add_request_calls) == 1) + stage1.push_engine_core_outputs(_terminal_engine_core_outputs(request_id)) + + data_msg = await _get_output_message(orchestrator_fixture) + terminal_msg = await _get_output_message(orchestrator_fixture) + + assert data_msg.engine_outputs is final_data + assert data_msg.finished is False + assert terminal_msg.request_id == request_id + assert terminal_msg.stage_id == 1 + assert terminal_msg.finished is True + assert terminal_msg.engine_outputs.finished is True + await _wait_for(lambda: request_id not in orchestrator_fixture.orchestrator.request_states) + await asyncio.sleep(0.05) + with pytest.raises(queue.Empty): + orchestrator_fixture.output_sync_q.get_nowait() + finally: + await _shutdown_orchestrator(orchestrator_fixture) + + +@pytest.mark.asyncio +async def test_async_chunk_processed_terminal_and_raw_terminal_finishes_once(orchestrator_factory) -> None: + request_id = "req-stream-processed-terminal" + stage0 = FakeStageClient(stage_type="llm", final_output=False) + stage1 = FakeStageClient(stage_type="llm", final_output=True, final_output_type="audio") + processed_terminal = _build_request_output( + request_id, + token_ids=[20, 21], + finished=True, + text="last audio chunk", + ) + orchestrator_fixture = orchestrator_factory( + [stage0, stage1], + output_processors=[ + FakeOutputProcessor(), + FakeOutputProcessor(request_outputs=[processed_terminal]), + ], + async_chunk=True, + ) + request = FakePromptRequest( + request_id=request_id, + prompt_token_ids=[1, 2, 3, 4], + ) + + try: + await _enqueue_add_request( + orchestrator_fixture, + request_id=request_id, + prompt=request, + original_prompt={"prompt": "stream audio"}, + sampling_params_list=[_sampling_params(), _sampling_params()], + final_stage_id=1, + ) + + await _wait_for(lambda: len(stage1.add_request_calls) == 1) + stage1.push_engine_core_outputs(_terminal_engine_core_outputs(request_id)) + + terminal_msg = await _get_output_message(orchestrator_fixture) + + assert terminal_msg.engine_outputs is processed_terminal + assert terminal_msg.finished is True + await _wait_for(lambda: request_id not in orchestrator_fixture.orchestrator.request_states) + await asyncio.sleep(0.05) + with pytest.raises(queue.Empty): + orchestrator_fixture.output_sync_q.get_nowait() + finally: + await _shutdown_orchestrator(orchestrator_fixture) + + @pytest.mark.asyncio async def test_run_shutdown(orchestrator_factory) -> None: stages = [ diff --git a/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py b/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py index 325daabdd..7664b2402 100644 --- a/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py +++ b/tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py @@ -1,5 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project """ E2E online tests for Qwen3-Omni /v1/realtime WebSocket (streaming PCM in, audio out). @@ -38,6 +38,10 @@ MODEL = "Qwen/Qwen3-Omni-30B-A3B-Instruct" REALTIME_SYNTH_PHRASE_TEXT = ( "Translate into Chinese: Beijing is the Capital of China. It is the center of culture and politics" ) +ISSUE_6474_SYNTH_PHRASE_TEXT = ( + "Can you tell me the current temperature and weather conditions in New York City? " + "What about Los Angeles? Please compare them in detail using at least eight complete sentences." +) # Simulate realtime upload pacing (``openai_realtime_client.py --send-delay-ms``). SEND_DELAY_MS = 200 @@ -100,6 +104,7 @@ async def _run_realtime_audio_roundtrip( *, chunk_ms: int = 100, send_delay_ms: int = 0, + completion_timeout_s: float = 600, ) -> dict: uri = f"ws://{host}:{port}/v1/realtime" incremental: list[bytes] = [] @@ -131,7 +136,7 @@ async def _run_realtime_audio_roundtrip( await ws.send(json.dumps({"type": "input_audio_buffer.commit", "final": True})) while True: - message = await asyncio.wait_for(ws.recv(), timeout=600) + message = await asyncio.wait_for(ws.recv(), timeout=completion_timeout_s) if isinstance(message, bytes): continue @@ -178,12 +183,16 @@ async def _run_realtime_audio_roundtrip( } -def _synthetic_pcm16_input() -> bytes: +def _synthetic_pcm16_input( + *, + phrase_text: str = REALTIME_SYNTH_PHRASE_TEXT, + duration_s: int = 10, +) -> bytes: syn = generate_synthetic_audio( - 10, + duration_s, 1, sample_rate=16000, - phrase_text=REALTIME_SYNTH_PHRASE_TEXT, + phrase_text=phrase_text, ) wav_bytes = base64.b64decode(syn["base64"]) return _pcm16_mono_16k_from_wav_bytes(wav_bytes) @@ -260,6 +269,33 @@ class TestQwen3OmniRealtimeWebSocket: _assert_realtime_smoke(result) _assert_realtime_accuracy(result) + @pytest.mark.advanced_model + @pytest.mark.omni + @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) + @pytest.mark.parametrize("omni_server", realtime_async_chunk_server_params, indirect=True) + def test_long_audio_response_completes_async_chunk(self, omni_server) -> None: + """Regression for #6474: long async-chunk audio must emit response.audio.done.""" + issue_6474_pcm16 = _synthetic_pcm16_input( + phrase_text=ISSUE_6474_SYNTH_PHRASE_TEXT, + ) + issue_6474_result = asyncio.run( + _run_realtime_audio_roundtrip( + omni_server.host, + omni_server.port, + omni_server.model, + issue_6474_pcm16, + chunk_ms=100, + send_delay_ms=SEND_DELAY_MS, + completion_timeout_s=180, + ) + ) + + _assert_realtime_smoke(issue_6474_result) + output_duration_s = len(issue_6474_result["output_pcm"]) / (2 * issue_6474_result["output_sample_rate"]) + assert output_duration_s > 5, ( + f"Expected an issue-like audio response longer than 5 seconds, got {output_duration_s:.2f}s" + ) + @pytest.mark.advanced_model @pytest.mark.omni @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2) diff --git a/vllm_omni/engine/orchestrator.py b/vllm_omni/engine/orchestrator.py index a9bc33118..e1306fe3e 100644 --- a/vllm_omni/engine/orchestrator.py +++ b/vllm_omni/engine/orchestrator.py @@ -978,6 +978,7 @@ class Orchestrator: for replica_id in pool.available_replica_ids(): if self._shutdown_event.is_set(): return + raw_terminal_request_ids: set[str] = set() # Shared catch so a dead replica on either poll path is # evicted rather than tearing down every stage (#4285). @@ -1038,8 +1039,8 @@ class Orchestrator: "new_prompt_len_snapshot", None, ) - if req_state.streaming.enabled: - await self._apply_raw_terminal_stage_finish(stage_id, eco, req_state) + if await self._apply_raw_terminal_stage_finish(stage_id, eco, req_state): + raw_terminal_request_ids.add(req_state.request_id) iteration_stats = ( IterationStats() if (self._stat_logger is not None and raw_outputs.outputs) else None ) @@ -1083,6 +1084,7 @@ class Orchestrator: raise await self._handle_processed_outputs(stage_id, replica_id, processed) + await self._finish_raw_terminal_requests(stage_id, replica_id, raw_terminal_request_ids) idle = False self._orch_monitor.note_loop(idle=idle) @@ -1434,8 +1436,8 @@ class Orchestrator: stage_id: int, eco: Any, req_state: OrchestratorRequestState, - ) -> None: - """Record session-level finish markers dropped by the streaming output processor. + ) -> bool: + """Record a session-level finish marker and report whether it was terminal. Streaming segment stops set ``is_segment_finished=True`` and are handled via processed outputs. Session termination (e.g. ``finish_requests`` after @@ -1448,14 +1450,60 @@ class Orchestrator: outputs after stage-0 session end. """ if getattr(eco, "finish_reason", None) is None: - return + return False if getattr(eco, "is_segment_finished", False): - return + return False final_output_stage_ids = req_state.final_output_stage_ids or {req_state.final_stage_id} if stage_id not in final_output_stage_ids: - return + return False req_state.finished_final_output_stage_ids.add(stage_id) + return True + + async def _finish_raw_terminal_requests( + self, + stage_id: int, + replica_id: int, + request_ids: set[str], + ) -> None: + """Finish streaming requests whose raw terminal had no processed output.""" + pool = self.stage_pools[stage_id] + if not pool.final_output: + return + + for request_id in request_ids: + req_state = self.request_states.get(request_id) + if req_state is None or self._is_duplex_session_request(req_state): + continue + + final_output_stage_ids = req_state.final_output_stage_ids or {req_state.final_stage_id} + if not final_output_stage_ids.issubset(req_state.finished_final_output_stage_ids): + continue + + final_output_type = getattr(pool.stage_client, "final_output_type", None) + logger.info( + "[Orchestrator] req=%s stage-%s raw terminal produced no processed terminal; returning empty %s output", + request_id, + stage_id, + final_output_type or "text", + ) + terminal_output = _build_terminal_empty_output( + request_id, + final_output_type=final_output_type, + audio_sample_rate=pool._infer_audio_sample_rate(), + ) + await self.output_async_queue.put( + OutputMessage( + request_id=request_id, + stage_id=stage_id, + replica_id=replica_id, + engine_outputs=terminal_output, + metrics=None, + finished=True, + stage_submit_ts=req_state.stage_submit_ts.get(stage_id), + ) + ) + await self._cleanup_request_ids([request_id, *self._cfg_tracker.cleanup_parent(request_id)]) def _maybe_clone_diffusion_params_for_cfg(self, request_id: str, params: Any) -> Any: """Attach CFG companion ids to diffusion sampling params when needed."""