[Bugfix] Fix missing Realtime audio completion (#6564)

Signed-off-by: psv666 <2693925048@qq.com>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>

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 = [

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)

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."""