diff --git a/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py b/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py index b6d562303..6a2809040 100644 --- a/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py +++ b/tests/e2e/online_serving/minicpmo_realtime_duplex_scenarios.py @@ -265,14 +265,26 @@ class DemoState: indices[event_type] = index return indices - def event_order_ok(self, *, require_input_commit: bool = True) -> bool: + def event_order_ok( + self, + *, + require_input_commit: bool = True, + require_commit_before_response: bool = True, + ) -> bool: if not self.events or self.events[0].get("type") != "session.created": return False first_commit_index = self.first_index("input_audio_buffer.committed") first_response_index = self.first_index("response.created") if first_response_index is None: return False - if require_input_commit and (first_commit_index is None or first_commit_index > first_response_index): + if require_input_commit and first_commit_index is None: + return False + if ( + require_input_commit + and require_commit_before_response + and first_commit_index is not None + and first_commit_index > first_response_index + ): return False indices_by_type = self.first_response_lifecycle_indices() if not indices_by_type: @@ -1409,7 +1421,10 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]: require_input_commit=not continuous_input, ) if validation_mode == "model-policy" - else state.event_order_ok(require_input_commit=not continuous_input) + else state.event_order_ok( + require_input_commit=not continuous_input, + require_commit_before_response=False, + ) ) input_transcription_ok = _input_transcription_ok( state.input_transcription_count, diff --git a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py index 2e409f689..895846147 100644 --- a/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py +++ b/tests/e2e/online_serving/test_minicpmo_4_5_duplex.py @@ -114,7 +114,7 @@ def test_duplex_single_session_response_required(omni_server, model_prefix: str, ) ) ) - assert result["ok"] is True + assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2) assert result["audio_delta_count"] > 0 assert result["done_count"] == 1 assert result["error_count"] == 0 @@ -137,7 +137,7 @@ def test_duplex_two_sessions_resume_and_takeover(omni_server, model_prefix: str, ) ) ) - assert result["ok"] is True + assert result["ok"] is True, json.dumps(result, ensure_ascii=False, indent=2) assert result["session_count"] == 2 assert result["resume"]["ok"] is True assert result["takeover"]["ok"] is True diff --git a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py index f21035f44..1dbfdbe2f 100644 --- a/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py +++ b/tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py @@ -874,6 +874,39 @@ def test_realtime_duplex_demo_model_policy_accepts_continuous_input_without_comm assert state.model_policy_event_order_ok(expected_turns=3, require_input_commit=False) +def test_realtime_duplex_demo_distinguishes_late_and_missing_commit(): + demo = _load_demo_module() + state = demo.DemoState() + response_id = "resp-1" + item_id = f"item_{response_id}" + for event in ( + {"type": "session.created"}, + {"type": "response.created", "response_id": response_id}, + {"type": "conversation.item.added", "item": {"id": item_id}}, + {"type": "response.output_item.added", "response_id": response_id}, + {"type": "response.content_part.added", "response_id": response_id}, + {"type": "response.speak", "response_id": response_id}, + {"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA"}, + {"type": "response.audio.done", "response_id": response_id}, + {"type": "response.content_part.done", "response_id": response_id}, + {"type": "response.output_item.done", "response_id": response_id}, + {"type": "response.done", "response_id": response_id}, + {"type": "input_audio_buffer.committed"}, + ): + state.add(event) + + assert not state.event_order_ok(require_input_commit=True) + assert state.event_order_ok( + require_input_commit=True, + require_commit_before_response=False, + ) + state.events.pop() + assert not state.event_order_ok( + require_input_commit=True, + require_commit_before_response=False, + ) + + def test_realtime_duplex_demo_accepts_overlap_unit_terminating_continuous_response(): demo = _load_demo_module() state = demo.DemoState() diff --git a/tests/engine/test_async_omni_engine_stage_init.py b/tests/engine/test_async_omni_engine_stage_init.py index 87edc594b..c366fb313 100644 --- a/tests/engine/test_async_omni_engine_stage_init.py +++ b/tests/engine/test_async_omni_engine_stage_init.py @@ -1,3 +1,4 @@ +import concurrent.futures import contextlib import importlib import os @@ -7,6 +8,7 @@ import types import pytest from vllm_omni.diffusion.data import AttentionConfig, AttentionSpec +from vllm_omni.engine import async_omni_engine as async_omni_engine_module from vllm_omni.engine.async_omni_engine import AsyncOmniEngine from vllm_omni.engine.stage_init_utils import ( LogicalStageInitPlan, @@ -19,6 +21,30 @@ from vllm_omni.engine.stage_runtime import StageRuntime pytestmark = [pytest.mark.core_model, pytest.mark.cpu] +def test_orchestrator_startup_timeout_warns_how_to_raise_limits(monkeypatch): + engine = object.__new__(AsyncOmniEngine) + engine.orchestrator_thread = types.SimpleNamespace(is_alive=lambda: True) + monkeypatch.setattr(engine, "_try_shutdown", lambda _message: None) + + ticks = iter((0.0, 1.0)) + monkeypatch.setattr(async_omni_engine_module.time, "monotonic", lambda: next(ticks)) + + warnings: list[tuple[object, ...]] = [] + monkeypatch.setattr( + async_omni_engine_module.logger, + "warning", + lambda *args: warnings.append(args), + ) + + with pytest.raises(TimeoutError, match="did not become ready within 1s"): + engine._wait_for_orchestrator_init(concurrent.futures.Future(), startup_timeout=1) + + assert len(warnings) == 1 + message = str(warnings[0][0]) + assert "--init-timeout" in message + assert "--stage-init-timeout" in message + + def _make_llm_metadata( stage_id: int, *, diff --git a/tests/entrypoints/openai_api/test_serving_speech.py b/tests/entrypoints/openai_api/test_serving_speech.py index b6c7c7c87..3739f8860 100644 --- a/tests/entrypoints/openai_api/test_serving_speech.py +++ b/tests/entrypoints/openai_api/test_serving_speech.py @@ -25,6 +25,7 @@ from vllm.entrypoints.openai.engine.protocol import ErrorInfo, ErrorResponse from vllm_omni.entrypoints.omni_base import OmniEngineDeadError from vllm_omni.entrypoints.openai import api_server as api_server_module +from vllm_omni.entrypoints.openai import serving_speech as serving_speech_module from vllm_omni.entrypoints.openai.audio_utils_mixin import AudioMixin from vllm_omni.entrypoints.openai.protocol.audio import ( BatchSpeechRequest, @@ -1390,6 +1391,20 @@ class TestTTSMethods: # Verify speakers are normalized to lowercase assert server.supported_speakers == {"ryan", "vivian", "aiden"} + def test_load_supported_speakers_skips_non_tts_omni_model( + self, + speech_server, + mocker: MockerFixture, + ): + warning = mocker.patch.object(serving_speech_module.logger, "warning") + speech_server.engine_client.model_config = SimpleNamespace( + hf_config=SimpleNamespace(tts_config=SimpleNamespace()) + ) + + assert speech_server._is_tts is False + assert speech_server._load_supported_speakers() == set() + warning.assert_not_called() + def test_load_supported_languages_from_config(self, speech_server): """Languages/dialects from codec_language_id are loaded title-cased; 'Auto' is added.""" speech_server._tts_model_type = "qwen3_tts" diff --git a/vllm_omni/engine/async_omni_engine.py b/vllm_omni/engine/async_omni_engine.py index 1fd747df5..e72fd01c9 100644 --- a/vllm_omni/engine/async_omni_engine.py +++ b/vllm_omni/engine/async_omni_engine.py @@ -465,6 +465,13 @@ class AsyncOmniEngine: while True: remaining = deadline - time.monotonic() if remaining <= 0: + logger.warning( + "[AsyncOmniEngine] Orchestrator startup timed out after %ss. " + "Multi-stage deployments that initialize stages sequentially on one device " + "or load checkpoints from slow storage may need larger --init-timeout and " + "--stage-init-timeout values.", + startup_timeout, + ) self._try_shutdown("[AsyncOmniEngine] Failed to cleanup after orchestrator startup timeout") raise TimeoutError(f"Orchestrator did not become ready within {startup_timeout}s") try: diff --git a/vllm_omni/entrypoints/openai/serving_speech.py b/vllm_omni/entrypoints/openai/serving_speech.py index 9cc6118e9..6cec05db8 100644 --- a/vllm_omni/entrypoints/openai/serving_speech.py +++ b/vllm_omni/entrypoints/openai/serving_speech.py @@ -832,7 +832,9 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): config = self.engine_client.model_config.hf_config.audio_config else: # Default is qwen3_tts path - config = self.engine_client.model_config.hf_config.talker_config + config = getattr(self.engine_client.model_config.hf_config, "talker_config", None) + if config is None: + return set() # Check for speakers in either spk_id or speaker_id for attr_name in ["spk_id", "speaker_id"]: