[ci] add merge/ready ci for audio realtime api (#4069)

Signed-off-by: CHEN <116010019@link.cuhk.edu.cn>

.buildkite/test-ready.yml

@@ -273,6 +273,42 @@ steps:
                       path: /mnt/hf-cache
                       type: DirectoryOrCreate
 
+      - label: "Omni · Qwen3-Omni Realtime WebSocket"
+        commands:
+          - |
+            timeout 30m bash -c "
+              export VLLM_WORKER_MULTIPROC_METHOD=spawn
+              pytest -s -v tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py::TestQwen3OmniRealtimeWebSocket::test_streaming_audio_input_pcm_output_async_chunk_smoke -m 'core_model' --run-level 'core_model'
+            "
+        agents:
+          queue: "mithril-h100-pool"
+        plugins:
+          - kubernetes:
+              podSpec:
+                containers:
+                  - image: 936637512419.dkr.ecr.us-west-2.amazonaws.com/vllm-ci-pull-through-cache/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
+                    resources:
+                      limits:
+                        nvidia.com/gpu: 2
+                    volumeMounts:
+                      - name: devshm
+                        mountPath: /dev/shm
+                      - name: hf-cache
+                        mountPath: /root/.cache/huggingface
+                    env:
+                      - name: HF_HOME
+                        value: /root/.cache/huggingface
+                nodeSelector:
+                  node.kubernetes.io/instance-type: gpu-h100-sxm
+                volumes:
+                  - name: devshm
+                    emptyDir:
+                      medium: Memory
+                  - name: hf-cache
+                    hostPath:
+                      path: /mnt/hf-cache
+                      type: DirectoryOrCreate
+
       - label: "Omni · VoxCPM2 Test"
         timeout_in_minutes: 20
         commands:

tests/entrypoints/openai_api/test_qwen3_omni_realtime_websocket.py

@@ -2,6 +2,11 @@
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 """
 E2E online tests for Qwen3-Omni /v1/realtime WebSocket (streaming PCM in, audio out).
+
+Three scenarios:
+- Ready CI: async_chunk on, smoke only (no send delay, no accuracy check).
+- Merge CI: async_chunk on + send delay, full accuracy check.
+- Merge CI: async_chunk off, no send delay, full accuracy check.
 """
 
 from __future__ import annotations
@@ -34,18 +39,32 @@ REALTIME_SYNTH_PHRASE_TEXT = (
     "Translate into Chinese: Beijing is the Capital of China. It is the center of culture and politics"
 )
 
-# The new-schema CI overlay bakes in async_chunk: False and covers CUDA/ROCm/XPU
-# via its ``platforms:`` section, so one path serves all three.
+# Simulate realtime upload pacing (``openai_realtime_client.py --send-delay-ms``).
+SEND_DELAY_MS = 200
+
+# CI overlay bakes in async_chunk: False and covers CUDA/ROCm/XPU via ``platforms:``.
 default_stage_config = get_deploy_config_path("ci/qwen3_omni_moe.yaml")
 
-realtime_server_params = [
+realtime_sync_server_params = [
     pytest.param(
         OmniServerParams(
             model=MODEL,
             stage_config_path=default_stage_config,
             use_stage_cli=True,
         ),
-        id="default",
+        id="sync",
+    ),
+]
+
+realtime_async_chunk_server_params = [
+    pytest.param(
+        OmniServerParams(
+            model=MODEL,
+            stage_config_path=default_stage_config,
+            use_stage_cli=True,
+            server_args=["--async-chunk"],
+        ),
+        id="async_chunk",
     ),
 ]
 
@@ -80,6 +99,7 @@ async def _run_realtime_audio_roundtrip(
     pcm16: bytes,
     *,
     chunk_ms: int = 100,
+    send_delay_ms: int = 0,
 ) -> dict:
     uri = f"ws://{host}:{port}/v1/realtime"
     incremental: list[bytes] = []
@@ -105,6 +125,8 @@ async def _run_realtime_audio_roundtrip(
                     }
                 )
             )
+            if send_delay_ms > 0:
+                await asyncio.sleep(send_delay_ms / 1000.0)
 
         await ws.send(json.dumps({"type": "input_audio_buffer.commit", "final": True}))
 
@@ -156,26 +178,49 @@ async def _run_realtime_audio_roundtrip(
     }
 
 
+def _synthetic_pcm16_input() -> bytes:
+    syn = generate_synthetic_audio(
+        10,
+        1,
+        sample_rate=16000,
+        phrase_text=REALTIME_SYNTH_PHRASE_TEXT,
+    )
+    wav_bytes = base64.b64decode(syn["base64"])
+    return _pcm16_mono_16k_from_wav_bytes(wav_bytes)
+
+
+def _assert_realtime_smoke(result: dict) -> None:
+    out_pcm = result["output_pcm"]
+    assert result["delta_events"] >= 1
+    assert out_pcm, "No output PCM from response.audio.delta"
+    assert len(out_pcm) % 2 == 0
+    assert len(out_pcm) >= 4096, "Output audio unexpectedly small"
+    assert result["output_sample_rate"] > 0
+
+
+def _assert_realtime_accuracy(result: dict) -> None:
+    final_text = (result["transcription_text"] or "").strip()
+    assert final_text, "Expected non-empty transcription (model text stream)"
+
+    wav_out = _wav_bytes_from_pcm16(result["output_pcm"], result["output_sample_rate"])
+    whisper_text = convert_audio_bytes_to_text(wav_out).strip()
+    assert whisper_text, "Whisper returned empty string for synthesized output audio"
+
+    sim = cosine_similarity_text(whisper_text.lower(), final_text.lower())
+    assert sim > 0.8, (
+        f"Output audio transcript should match model text (sim={sim:.3f}): "
+        f"whisper={whisper_text!r}, model_text={final_text!r}"
+    )
+
+
 class TestQwen3OmniRealtimeWebSocket:
-    @pytest.mark.advanced_model
+    @pytest.mark.core_model
     @pytest.mark.omni
     @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2)
-    @pytest.mark.parametrize("omni_server", realtime_server_params, indirect=True)
-    def test_streaming_audio_input_pcm_output(self, omni_server) -> None:
-        """
-        Short streamed 16 kHz mono PCM16 input; expect streamed PCM16 audio deltas and
-        transcription. Verify Whisper(output audio) aligns with model text (same idea
-        as multimodal omni e2e). Input speech is synthesized from
-        ``REALTIME_SYNTH_PHRASE_TEXT``.
-        """
-        syn = generate_synthetic_audio(
-            10,
-            1,
-            sample_rate=16000,
-            phrase_text=REALTIME_SYNTH_PHRASE_TEXT,
-        )
-        wav_bytes = base64.b64decode(syn["base64"])
-        pcm16 = _pcm16_mono_16k_from_wav_bytes(wav_bytes)
+    @pytest.mark.parametrize("omni_server", realtime_async_chunk_server_params, indirect=True)
+    def test_streaming_audio_input_pcm_output_async_chunk_smoke(self, omni_server) -> None:
+        """Ready CI: async_chunk on, no send delay, smoke only."""
+        pcm16 = _synthetic_pcm16_input()
 
         result = asyncio.run(
             _run_realtime_audio_roundtrip(
@@ -184,25 +229,52 @@ class TestQwen3OmniRealtimeWebSocket:
                 omni_server.model,
                 pcm16,
                 chunk_ms=100,
+                send_delay_ms=0,
             )
         )
 
-        out_pcm = result["output_pcm"]
-        assert result["delta_events"] >= 1
-        assert out_pcm, "No output PCM from response.audio.delta"
-        assert len(out_pcm) % 2 == 0
-        assert len(out_pcm) >= 4096, "Output audio unexpectedly small"
-        assert result["output_sample_rate"] > 0
+        _assert_realtime_smoke(result)
 
-        final_text = (result["transcription_text"] or "").strip()
-        assert final_text, "Expected non-empty transcription (model text stream)"
+    @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_streaming_audio_input_pcm_output_async_chunk(self, omni_server) -> None:
+        """Merge CI: async_chunk on, paced upload, full accuracy check."""
+        pcm16 = _synthetic_pcm16_input()
 
-        wav_out = _wav_bytes_from_pcm16(out_pcm, result["output_sample_rate"])
-        whisper_text = convert_audio_bytes_to_text(wav_out).strip()
-        assert whisper_text, "Whisper returned empty string for synthesized output audio"
-
-        sim = cosine_similarity_text(whisper_text.lower(), final_text.lower())
-        assert sim > 0.8, (
-            f"Output audio transcript should match model text (sim={sim:.3f}): "
-            f"whisper={whisper_text!r}, model_text={final_text!r}"
+        result = asyncio.run(
+            _run_realtime_audio_roundtrip(
+                omni_server.host,
+                omni_server.port,
+                omni_server.model,
+                pcm16,
+                chunk_ms=100,
+                send_delay_ms=SEND_DELAY_MS,
+            )
         )
+
+        _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_sync_server_params, indirect=True)
+    def test_streaming_audio_input_pcm_output(self, omni_server) -> None:
+        """Merge CI: async_chunk off, no send delay, full accuracy check."""
+        pcm16 = _synthetic_pcm16_input()
+
+        result = asyncio.run(
+            _run_realtime_audio_roundtrip(
+                omni_server.host,
+                omni_server.port,
+                omni_server.model,
+                pcm16,
+                chunk_ms=100,
+                send_delay_ms=0,
+            )
+        )
+
+        _assert_realtime_smoke(result)
+        _assert_realtime_accuracy(result)