[Test] Match invalid realtime requests to the current endpoint contract

Signed-off-by: tzhouam <tzhouam@connect.ust.hk>

.buildkite/cuda/rebase-pipeline.yml

@@ -1,46 +1,35 @@
-# Focused weekly comparison against unchanged Omni main and official vLLM 0.30.0.
 steps:
-- label: ':docker: Build image'
-  key: image-build
-  commands:
-  - bash .buildkite/cuda/scripts/build-ci-image.sh
-  agents:
-    queue: cpu_queue_premerge
-- label: Upload selected weekly baseline
-  depends_on: image-build
-  key: upload-weekly-pipeline
-  commands:
-  - |
-    python3 - <<'PYBASELINE'
-    import tempfile
-    from pathlib import Path
-    import yaml
-    labels = set(['TTS · Audex · CosyVoice · Qwen3-TTS · L4', 'TTS · H100', 'Diffusion · Bagel · H100 · Single-GPU', 'Diffusion · SenseNova-U1 · H100 · Single-GPU', 'Diffusion · DreamZero · H100 · 2-GPU', 'Diffusion · H100 · 4-GPU', 'Diffusion · MAGI-2 Preview Serving Smoke · H100 · 4-GPU'])
-    source = yaml.safe_load(Path('.buildkite/cuda/test-weekly.yml').read_text())
-    selected = set()
-    guard = 'python3 -c \'import vllm, torch, flashinfer, vllm._C_stable_libtorch; assert vllm.__version__ == \'"\'"\'0.30.0\'"\'"\'; assert torch.cuda.is_available(); print(\'"\'"\'Verified baseline runtime:\'"\'"\', vllm.__version__, torch.__version__)\''
-    def select(steps):
-        kept = []
-        for step in steps:
-            if not isinstance(step, dict):
-                continue
-            if 'steps' in step:
-                children = select(step['steps'])
-                if children:
-                    kept.append(dict(step, steps=children))
-            elif step.get('label') in labels:
-                item = dict(step)
-                commands = item['commands']
-                item['commands'] = [guard] + ([commands] if isinstance(commands, str) else commands)
-                kept.append(item)
-                selected.add(item['label'])
-        return kept
-    source['steps'] = select(source['steps'])
-    assert selected == labels, (selected, labels)
-    path = Path(tempfile.gettempdir()) / 'vllm-omni-selected-weekly-baseline.yml'
-    path.write_text(yaml.safe_dump(source, sort_keys=False, allow_unicode=True))
-    print('Selected baseline suites:', sorted(selected), flush=True)
-    PYBASELINE
-    python3 .buildkite/common/scripts/upload_pipeline.py --upload --all /tmp/vllm-omni-selected-weekly-baseline.yml
-  agents:
-    queue: cpu_queue_premerge
+  - label: ":docker: Build image"
+    key: image-build
+    commands:
+      - "bash .buildkite/cuda/scripts/build-ci-image.sh"
+    agents:
+      queue: "cpu_queue_premerge"
+
+  # L2 Test
+  - label: "Upload Ready Pipeline"
+    depends_on: image-build
+    key: upload-ready-pipeline
+    commands:
+      - python3 .buildkite/common/scripts/upload_pipeline.py --upload --all .buildkite/cuda/test-ready.yml
+    agents:
+      queue: "cpu_queue_premerge"
+
+  # L3 Test
+  - label: "Upload Merge Pipeline"
+    depends_on: image-build
+    key: upload-merge-pipeline
+    commands:
+      - python3 .buildkite/common/scripts/upload_pipeline.py --upload --all .buildkite/cuda/test-merge.yml
+    agents:
+      queue: "cpu_queue_premerge"
+
+  # L4 Test
+  - label: "Upload Nightly Pipeline"
+    depends_on: image-build
+    key: upload-nightly-pipeline
+    if: build.env("NIGHTLY") == "1"
+    commands:
+      - python3 .buildkite/common/scripts/upload_pipeline.py --upload --all .buildkite/cuda/test-nightly.yml
+    agents:
+      queue: "cpu_queue_premerge"

docker/Dockerfile.ci

@@ -36,8 +36,7 @@ RUN apt-get update && \
 # for CI
 COPY pyproject.toml setup.py ./
 COPY requirements/ requirements/
-# Keep source-built dependency versions independent of the Omni CI version.
-ENV VLLM_OMNI_VERSION_OVERRIDE=0.0.0
+ENV SETUPTOOLS_SCM_PRETEND_VERSION=0.0.0
 # Editable install: deps + project metadata are cached in this layer, while
 # the package itself is linked to APP_DIR (populated later by COPY --link).
 # Do NOT use a non-editable install here — that copies an empty stub into

tests/dfx/reliability/invalid_param_test/test_invalid_omni_chat.py

@@ -216,44 +216,40 @@ def test_video_chat_stream_invalid_requests(
 # WS /v1/realtime
 # ─────────────────────────────────────────────────────────────────────────────
 
-# Resolved in ``test_realtime_invalid_requests`` (skip ``session.created`` handshake).
-_REALTIME_WS_MODEL_MISMATCH = object()
-_REALTIME_WS_INVALID_AUDIO_APPEND = object()
-
 
 @hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2)
 @pytest.mark.parametrize(
     "send_frames_spec, ws_error_code, err_message",
     [
-        pytest.param("{not-json", "invalid_json", "Invalid JSON", id="invalid_json"),
+        pytest.param("{not-json", "invalid_event", "Invalid or unrecognized client event", id="invalid_json"),
         pytest.param(
             json.dumps({"type": "session.update"}),
             "invalid_event",
-            ("model", "Missing required field"),
-            id="session_update_missing_model",
+            "Invalid or unrecognized client event",
+            id="session_update_missing_session",
         ),
         pytest.param(
-            _REALTIME_WS_MODEL_MISMATCH,
-            "model_not_found",
-            ("model", "does not exist"),
-            id="session_update_model_mismatch",
+            json.dumps({"type": "session.update", "session": []}),
+            "invalid_event",
+            "Invalid or unrecognized client event",
+            id="session_update_invalid_session",
         ),
         pytest.param(
-            json.dumps({"type": "input_audio_buffer.commit", "final": False}),
-            "model_not_validated",
-            ("session.update", "validate the model"),
-            id="commit_without_model_validation",
+            json.dumps({"type": "input_audio_buffer.commit"}),
+            "invalid_request_error",
+            "Input audio buffer is empty",
+            id="commit_without_audio",
         ),
         pytest.param(
             json.dumps({"type": "unknown.event"}),
-            "unknown_event",
-            "Unknown event type",
+            "invalid_event",
+            "Invalid or unrecognized client event",
             id="unknown_event_type",
         ),
         pytest.param(
-            _REALTIME_WS_INVALID_AUDIO_APPEND,
-            "invalid_audio",
-            "Invalid audio data",
+            json.dumps({"type": "input_audio_buffer.append", "audio": "not-valid-base64!!!"}),
+            "invalid_request_error",
+            "Invalid base64 audio data",
             id="invalid_audio_append",
         ),
     ],
@@ -266,27 +262,12 @@ def test_realtime_invalid_requests(
     ws_error_code: str,
     err_message: str | tuple[str, ...],
 ) -> None:
-    send_frames: Any
-    if send_frames_spec is _REALTIME_WS_MODEL_MISMATCH:
-        send_frames = json.dumps(
-            {
-                "type": "session.update",
-                "model": "this-model-is-not-served-on-this-instance",
-            }
-        )
-    elif send_frames_spec is _REALTIME_WS_INVALID_AUDIO_APPEND:
-        send_frames = [
-            json.dumps({"type": "session.update", "model": omni_server.model}),
-            json.dumps({"type": "input_audio_buffer.append", "audio": "not-valid-base64!!!"}),
-        ]
-    else:
-        send_frames = send_frames_spec
     online_client.send_realtime_ws_request(
         {
-            "send_frames": send_frames,
+            "send_frames": send_frames_spec,
             "timeout": 120,
             "ws_max_size": None,
-            "ws_skip_types": ["session.created"],
+            "ws_skip_types": ["session.created", "conversation.created"],
             "ws_json_type": "error",
             "ws_error_code": ws_error_code,
             "err_message": err_message,

tests/entrypoints/test_realtime_connection_helpers.py

@@ -4,7 +4,10 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
+import json
+from types import SimpleNamespace
 
 import numpy as np
 import pytest
@@ -84,3 +87,56 @@ class TestAsyncOmniStreamingParamsValidation:
         p = SamplingParams(n=1, stop=["\n"], output_kind=RequestOutputKind.DELTA)
         with pytest.raises(ValueError, match="Input streaming"):
             AsyncOmni._validate_streaming_input_sampling_params(p)
+
+
+@pytest.mark.parametrize(
+    "frame,code,message",
+    [
+        ("{not-json", "invalid_event", "Invalid or unrecognized client event"),
+        (json.dumps({"type": "session.update"}), "invalid_event", "Invalid or unrecognized client event"),
+        (
+            json.dumps({"type": "session.update", "session": []}),
+            "invalid_event",
+            "Invalid or unrecognized client event",
+        ),
+        (json.dumps({"type": "input_audio_buffer.commit"}), "invalid_request_error", "Input audio buffer is empty"),
+        (json.dumps({"type": "unknown.event"}), "invalid_event", "Invalid or unrecognized client event"),
+        (
+            json.dumps({"type": "input_audio_buffer.append", "audio": "not-valid-base64!!!"}),
+            "invalid_request_error",
+            "Invalid base64 audio data",
+        ),
+    ],
+)
+def test_full_duplex_invalid_request_contract(frame, code, message):
+    from starlette.websockets import WebSocketDisconnect
+
+    from vllm_omni.entrypoints.openai.realtime.connection import OpenAIFullDuplexConnection
+
+    class Socket:
+        def __init__(self):
+            self.sent = []
+            self.frames = iter([frame])
+
+        async def accept(self):
+            pass
+
+        async def receive_text(self):
+            value = next(self.frames, None)
+            if value is None:
+                raise WebSocketDisconnect()
+            return value
+
+        async def send_text(self, payload):
+            self.sent.append(json.loads(payload))
+
+    async def run():
+        socket = Socket()
+        connection = OpenAIFullDuplexConnection(socket, SimpleNamespace(), "test-model", None)
+        await connection.handle_connection()
+        assert [event["type"] for event in socket.sent] == ["session.created", "conversation.created", "error"]
+        error = socket.sent[-1]["error"]
+        assert error["code"] == code
+        assert error["message"] == message
+
+    asyncio.run(run())