[Frontend] Add a shared realtime web UI for MiniCPM-o and Qwen3-Omni (#7222) (#7585)

Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>

tests/entrypoints/openai_api/test_api_server_guards.py

@@ -47,6 +47,7 @@ from argparse import Namespace
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from types import SimpleNamespace
+from typing import Any
 
 import pytest
 from fastapi import FastAPI, HTTPException
@@ -564,7 +565,7 @@ async def test_timestamp_middleware_stamps_http_and_passes_websocket(monkeypatch
     Fails if HTTP requests lose ``request_timestamp``, or WebSocket scopes
     start getting stamped / mutated incorrectly.
     """
-    captured: dict[str, object] = {}
+    captured: dict[str, Any] = {}
 
     def fake_build_openai_app(args, supported_tasks):
         return FastAPI()
@@ -592,8 +593,8 @@ async def test_timestamp_middleware_stamps_http_and_passes_websocket(monkeypatch
     await api_server.omni_run_server_worker("127.0.0.1:8000", _FakeSocket(), _minimal_args())
     middleware = captured["served_app"]
 
-    http_scope = {"type": "http", "state": {}}
-    ws_scope = {"type": "websocket"}
+    http_scope: dict[str, Any] = {"type": "http", "state": {}}
+    ws_scope: dict[str, Any] = {"type": "websocket"}
 
     async def _receive():
         return {"type": "http.disconnect"}
@@ -920,11 +921,10 @@ async def test_pure_diffusion_app_state_key_snapshot(monkeypatch) -> None:
     engine = _FakeEngineClient(stage_configs=[stage])
 
     def _for_diffusion_factory(label: str):
-        @classmethod
         def _factory(cls, *args, **kwargs):
             return _marker(label)
 
-        return _factory
+        return classmethod(_factory)
 
     monkeypatch.setattr(api_server.OmniOpenAIServingChat, "for_diffusion", _for_diffusion_factory("chat"))
     monkeypatch.setattr(api_server.OmniOpenAIServingChatBatch, "for_diffusion", _for_diffusion_factory("chat_batch"))
@@ -965,13 +965,11 @@ async def test_pure_diffusion_speech_forwards_media_access_args(monkeypatch) ->
     speech_kwargs = {}
 
     def _for_diffusion_factory(label: str):
-        @classmethod
         def _factory(cls, *args, **kwargs):
             return _marker(label)
 
-        return _factory
+        return classmethod(_factory)
 
-    @classmethod
     def _speech_factory(cls, *args, **kwargs):
         speech_kwargs.update(kwargs)
         return _marker("speech")
@@ -985,7 +983,7 @@ async def test_pure_diffusion_speech_forwards_media_access_args(monkeypatch) ->
     )
     monkeypatch.setattr(api_server.OmniOpenAIServingVideo, "for_diffusion", _for_diffusion_factory("video"))
     monkeypatch.setattr(api_server.OmniStreamingVideoOutputHandler, "__init__", lambda self, *a, **k: None)
-    monkeypatch.setattr(api_server.OmniOpenAIServingSpeech, "for_diffusion", _speech_factory)
+    monkeypatch.setattr(api_server.OmniOpenAIServingSpeech, "for_diffusion", classmethod(_speech_factory))
     monkeypatch.setattr(
         api_server.ServingRealtimeRobotOpenPI,
         "create_policy_server",

tests/entrypoints/openai_api/test_duplex_api_server.py

@@ -490,3 +490,31 @@ async def test_a_model_that_does_not_declare_chat_completions_does_not_get_the_r
 
     assert state.openai_serving_chat is None
     assert isinstance(state.openai_serving_duplex, OmniDuplexSessionHandler)
+
+
+def test_qwen_plugin_preserves_custom_turn_deployment_default(monkeypatch, tmp_path):
+    from vllm_omni.model_executor.models.qwen3_omni.pipeline import QWEN3_OMNI_PIPELINE
+
+    monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", lambda **kwargs: QWEN3_OMNI_PIPELINE)
+    deploy = tmp_path / "qwen.yaml"
+    deploy.write_text("stages: []\n")
+    assert not api_server._should_serve_duplex("qwen", {"deploy_config": str(deploy)})
+    deploy.write_text("session_mode: duplex\nstages: []\n")
+    assert api_server._should_serve_duplex("qwen", {"deploy_config": str(deploy)})
+
+
+@pytest.mark.parametrize("flag", ["1", "true", "on"])
+def test_turn_deployment_rejects_explicit_duplex_without_falling_back_to_stt(flag: str) -> None:
+    app = _duplex_app(None)
+    # Even an available STT service must not silently accept a duplex request.
+    app.state.openai_serving_realtime = object()
+    with TestClient(app) as client:
+        with client.websocket_connect(f"/v1/realtime?duplex={flag}") as websocket:
+            assert websocket.receive_json() == {
+                "type": "error",
+                "code": "unsupported",
+                "error": "VAD realtime is not enabled",
+            }
+            with pytest.raises(WebSocketDisconnect) as exc:
+                websocket.receive_text()
+            assert exc.value.code == 1008