diff --git a/tests/entrypoints/openai_api/test_duplex_api_server.py b/tests/entrypoints/openai_api/test_duplex_api_server.py
index 0d9eebe5b..31421f6ff 100644
--- a/tests/entrypoints/openai_api/test_duplex_api_server.py
+++ b/tests/entrypoints/openai_api/test_duplex_api_server.py
@@ -14,10 +14,12 @@ from fastapi import FastAPI
 from fastapi.testclient import TestClient
 from starlette.datastructures import State
 from starlette.websockets import WebSocketDisconnect
+from vllm.entrypoints.openai.models.protocol import BaseModelPath
 
 from vllm_omni.config import stage_config
 from vllm_omni.config.config_factory import StageConfigFactory
-from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig, PipelineConfig, StagePipelineConfig
+from vllm_omni.config.omni_config import VllmOmniConfig
+from vllm_omni.config.stage_config import DeployConfig, DuplexSessionRuntimeConfig, PipelineConfig, StagePipelineConfig
 from vllm_omni.engine.duplex.config import DuplexCapabilities
 from vllm_omni.entrypoints.duplex.serving import OmniDuplexSessionHandler
 from vllm_omni.entrypoints.duplex_omni import DuplexOmni
@@ -438,6 +440,58 @@ def test_realtime_opted_out_of_duplex_falls_through_to_the_turn_based_route(quer
     assert handler.queries == []
 
 
+@pytest.fixture
+def qwen3_realtime_stages():
+    from vllm_omni.model_executor.models.qwen3_omni.pipeline import QWEN3_OMNI_PIPELINE
+
+    return VllmOmniConfig.from_pipeline_config(
+        QWEN3_OMNI_PIPELINE, user_deploy_config=DeployConfig(async_chunk=True)
+    ).stage_configs
+
+
+def test_qwen3_realtime_recognizes_typed_stages(qwen3_realtime_stages) -> None:
+    from vllm_omni.entrypoints.duplex.openai import supports_qwen3_omni_realtime
+
+    assert supports_qwen3_omni_realtime(qwen3_realtime_stages)
+
+
+@pytest.mark.parametrize("topology", ["missing-stage", "duplicate-stage", "other-architecture"])
+def test_qwen3_realtime_rejects_other_topologies(qwen3_realtime_stages, topology: str) -> None:
+    from vllm_omni.entrypoints.duplex.openai import supports_qwen3_omni_realtime
+
+    stages = list(qwen3_realtime_stages)
+    if topology == "missing-stage":
+        stages.pop()
+    elif topology == "duplicate-stage":
+        stages[-1] = stages[0]
+    else:
+        stages[0].model_config.model_arch = "Qwen2_5OmniForConditionalGeneration"
+
+    assert not supports_qwen3_omni_realtime(stages)
+
+
+def test_qwen3_turn_deployment_uses_conformant_realtime(qwen3_realtime_stages, mocker) -> None:
+    app = _duplex_app(None)
+    app.state.stage_configs = qwen3_realtime_stages
+    app.state.args = Namespace()
+    app.state.openai_serving_models = _FakeModels(
+        base_model_paths=[BaseModelPath(name="qwen3-omni", model_path="qwen3-omni")]
+    )
+    app.state.engine_client = mocker.Mock(
+        get_tokenizer=mocker.AsyncMock(return_value=mocker.Mock(chat_template="{{ messages }}"))
+    )
+
+    with TestClient(app) as client:
+        with client.websocket_connect("/v1/realtime?model=qwen3-omni") as websocket:
+            assert websocket.receive_json()["type"] == "session.created"
+            assert websocket.receive_json()["type"] == "conversation.created"
+            websocket.send_json({"type": "unknown_event"})
+            error = websocket.receive_json()
+            assert error["type"] == "error"
+            assert error["error"]["type"] == "invalid_request_error"
+            assert error["error"]["code"] == "invalid_event"
+
+
 # --------------------------------------------------------------------------- #
 # Startup warmup gate                                                         #
 # --------------------------------------------------------------------------- #
diff --git a/vllm_omni/entrypoints/duplex/openai.py b/vllm_omni/entrypoints/duplex/openai.py
index e55d1118c..946c489bd 100644
--- a/vllm_omni/entrypoints/duplex/openai.py
+++ b/vllm_omni/entrypoints/duplex/openai.py
@@ -4,8 +4,8 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Mapping
-from typing import Any
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
 
 from fastapi import WebSocket
 from vllm.logger import init_logger
@@ -15,22 +15,21 @@ from vllm_omni.entrypoints.duplex.warmup import DUPLEX_WARMUP_CLIENT_WAIT_S
 from vllm_omni.entrypoints.openai.realtime.connection import OpenAIFullDuplexConnection
 from vllm_omni.entrypoints.openai.realtime_connection import RealtimeConnection
 
+if TYPE_CHECKING:
+    from vllm_omni.config.omni_config import BaseVllmOmniStageConfig
+
 logger = init_logger(__name__)
 
 _QWEN3_OMNI_REALTIME_ARCH = "Qwen3OmniMoeForConditionalGeneration"
 _QWEN3_OMNI_REALTIME_STAGES = {"thinker", "talker", "code2wav"}
 
 
-def supports_qwen3_omni_realtime(stage_configs: Any) -> bool:
-    def stage_arg(stage: Any, name: str) -> Any:
-        engine_args = getattr(stage, "engine_args", None)
-        return engine_args.get(name) if isinstance(engine_args, Mapping) else getattr(engine_args, name, None)
-
+def supports_qwen3_omni_realtime(stage_configs: Sequence[BaseVllmOmniStageConfig] | None) -> bool:
     stages = stage_configs or ()
     return (
         len(stages) == len(_QWEN3_OMNI_REALTIME_STAGES)
-        and all(stage_arg(stage, "model_arch") == _QWEN3_OMNI_REALTIME_ARCH for stage in stages)
-        and {stage_arg(stage, "model_stage") for stage in stages} == _QWEN3_OMNI_REALTIME_STAGES
+        and all(stage.model_config.model_arch == _QWEN3_OMNI_REALTIME_ARCH for stage in stages)
+        and {stage.model_stage for stage in stages} == _QWEN3_OMNI_REALTIME_STAGES
     )