[CI] invalid_param reliability suite and weekly http_invalid jobs (#3652)

Signed-off-by: wangyu <410167048@qq.com>

.buildkite/test-weekly.yml

@@ -84,3 +84,65 @@ steps:
                 hostPath:
                   path: /mnt/hf-cache
                   type: DirectoryOrCreate
+
+  # Invalid HTTP / WS payloads vs live Omni servers (Issue #3649 tracks known skips).
+  - group: "Reliability Test - Invalid parameters Test"
+    depends_on: upload-weekly-pipeline
+    if: build.env("WEEKLY") == "1" || build.pull_request.labels includes "weekly-test"
+    steps:
+      - label: "Invalid parameters Test · H100"
+        timeout_in_minutes: 120
+        commands:
+          - pytest -s -v tests/dfx/reliability/invalid_param_test/ -m "slow and H100"
+        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
+                      - name: HF_TOKEN
+                        valueFrom:
+                          secretKeyRef:
+                            name: hf-token-secret
+                            key: token
+                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
+
+      # Stable Audio Open, Qwen3-TTS speech/batch/voices paths marked ``L4``.
+      - label: "Invalid parameters Test · L4"
+        timeout_in_minutes: 120
+        commands:
+          - pytest -s -v tests/dfx/reliability/invalid_param_test/ -m "slow and L4"
+        agents:
+          queue: "gpu_1_queue"
+        plugins:
+          - docker#v5.2.0:
+              image: public.ecr.aws/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
+              always-pull: true
+              propagate-environment: true
+              shm-size: "8gb"
+              environment:
+                - "HF_HOME=/fsx/hf_cache"
+                - "HF_TOKEN"
+              volumes:
+                - "/fsx/hf_cache:/fsx/hf_cache"

tests/conftest.py

@@ -36,6 +36,7 @@ def pytest_terminal_summary(terminalreporter, exitstatus, config):
 _ASSERTION_EXPORT_NAMES = (
     "assert_audio_speech_response",
     "assert_diffusion_response",
+    "assert_http_error",
     "assert_image_diffusion_response",
     "assert_image_valid",
     "assert_omni_response",
@@ -54,6 +55,7 @@ _MEDIA_EXPORT_NAMES = (
 _STAGE_CONFIG_EXPORT_NAMES = ("modify_stage_config",)
 _RUNTIME_EXPORT_NAMES = (
     "DiffusionResponse",
+    "HttpResponse",
     "OmniResponse",
     "OmniRunner",
     "OmniRunnerHandler",

tests/dfx/reliability/invalid_param_test/conftest.py

@@ -0,0 +1,23 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Shared env for ``http_invalid`` (real :func:`omni_server`)."""
+
+from __future__ import annotations
+
+import io
+import os
+
+import pytest
+from PIL import Image
+
+# Match ``tests/e2e/online_serving/*`` module-level env for subprocess serve.
+os.environ.setdefault("VLLM_WORKER_MULTIPROC_METHOD", "spawn")
+os.environ.setdefault("VLLM_TEST_CLEAN_GPU_MEMORY", "0")
+
+
+@pytest.fixture
+def tiny_png_bytes() -> bytes:
+    img = Image.new("RGB", (32, 32), color="gray")
+    buf = io.BytesIO()
+    img.save(buf, format="PNG")
+    return buf.getvalue()

tests/dfx/reliability/invalid_param_test/test_invalid_audio_diffusion.py

@@ -0,0 +1,118 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""``POST /v1/audio/generate`` validation (live Stable Audio Open server)."""
+
+from __future__ import annotations
+
+import pytest
+
+from tests.helpers.mark import hardware_marks
+from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler
+
+pytestmark = [pytest.mark.slow, pytest.mark.diffusion]
+
+# https://github.com/vllm-project/vllm-omni/issues/3649 — invalid inputs still HTTP 200 / timeout / missing validation.
+_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649")
+
+_PARAMS = [
+    pytest.param(
+        OmniServerParams(
+            model="stabilityai/stable-audio-open-1.0",
+            server_args=[
+                "--trust-remote-code",
+                "--enforce-eager",
+                "--model-class-name",
+                "StableAudioPipeline",
+            ],
+        ),
+        id="stable_audio_open",
+        marks=hardware_marks(res={"cuda": "L4"}),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "overrides, err_message",
+    [
+        pytest.param({"input": ""}, ("cannot be empty", "input"), id="input_empty", marks=_SKIP_ISSUE_3649),
+        pytest.param(
+            {"input": "   "}, ("cannot be empty", "input"), id="input_whitespace_only", marks=_SKIP_ISSUE_3649
+        ),
+        pytest.param({"input": 123}, ("input", "a valid string"), id="input_wrong_type"),
+        pytest.param(
+            {"response_format": "mpeg"},
+            ("response_format", "literal_error", "Input should be"),
+            id="response_format_invalid",
+        ),
+        pytest.param({"speed": 0.1}, ("speed", "greater_than_equal", "0.25"), id="speed_too_low"),
+        pytest.param({"speed": 99.0}, ("speed", "less_than_equal", "4"), id="speed_too_high"),
+        pytest.param({"stream_format": 1}, ("stream_format", "literal_error", "audio"), id="stream_format_wrong_type"),
+        pytest.param(
+            {"stream_format": "sse"},
+            ("sse", "stream_format", "value_error", "not a supported stream_format"),
+            id="stream_format_sse_blocked",
+        ),
+        pytest.param(
+            {"audio_length": -1.0}, ("audio_end_in_s", "higher than", "audio_start_in_s"), id="audio_length_negative"
+        ),
+        pytest.param({"audio_length": 0}, "audio_length", id="audio_length_zero", marks=_SKIP_ISSUE_3649),
+        # ``OmniOpenAIServingAudioGenerate`` only forwards ``audio_start`` when ``audio_length`` is set.
+        pytest.param(
+            {"audio_start": -0.5, "audio_length": 5.0},
+            ("audio_start_in_s", "between", "0", "512"),
+            id="audio_start_negative",
+        ),
+        pytest.param(
+            {"negative_prompt": ["noise"]}, ("negative_prompt", "a valid string"), id="negative_prompt_wrong_type"
+        ),
+        pytest.param({"guidance_scale": -1.0}, "guidance_scale", id="guidance_scale_negative", marks=_SKIP_ISSUE_3649),
+        pytest.param({"guidance_scale": 0}, "guidance_scale", id="guidance_scale_zero", marks=_SKIP_ISSUE_3649),
+        pytest.param(
+            {"num_inference_steps": 0}, "num_inference_steps", id="num_inference_steps_zero", marks=_SKIP_ISSUE_3649
+        ),
+        pytest.param(
+            {"num_inference_steps": -1}, ("number of steps", "non-negative"), id="num_inference_steps_negative"
+        ),
+        pytest.param(
+            {"num_inference_steps": 6000},
+            "num_inference_steps",
+            id="num_inference_steps_above_max",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            {"guidance_scale": 1001.0}, "guidance_scale", id="guidance_scale_above_max", marks=_SKIP_ISSUE_3649
+        ),
+        pytest.param(
+            {"audio_length": 86401.0}, ("Requested audio length", "exceeds maximum"), id="audio_length_above_max"
+        ),
+        pytest.param(
+            {"audio_start": 86401.0, "audio_length": 1.0},
+            ("audio_start_in_s", "between", "0", "512"),
+            id="audio_start_above_max",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True)
+def test_audio_generate_invalid_field_values(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    overrides: dict[str, object],
+    err_message: str | tuple[str, ...],
+) -> None:
+    body = {"model": omni_server.model, "input": "ambient electronic pad"}
+    body.update(overrides)
+    openai_client.send_audio_generate_http_request(
+        {"json": body, "timeout": 120, "err_code": 400, "err_message": err_message}
+    )
+
+
+@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True)
+def test_audio_generate_missing_input(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_audio_generate_http_request(
+        {
+            "json": {"model": omni_server.model},
+            "timeout": 120,
+            "err_code": 400,
+            "err_message": ("missing", "input", "field required"),
+        }
+    )

tests/dfx/reliability/invalid_param_test/test_invalid_audio_speech.py

@@ -0,0 +1,862 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""HTTP + WebSocket validation for Qwen3-TTS Base: ``/v1/audio/speech``, ``/v1/audio/speech/stream``, batch, voices."""
+
+from __future__ import annotations
+
+import io
+import json
+import uuid
+import wave
+from typing import Any
+
+import pytest
+
+from tests.helpers.mark import hardware_marks
+from tests.helpers.media import load_test_audio_data_url
+from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler
+from tests.helpers.stage_config import get_deploy_config_path
+
+# Speech / voice-upload numeric caps for these tests only (not exported from vllm_omni).
+# Align where possible with serving_speech defaults: _TTS_MAX_INSTRUCTIONS_LENGTH (500),
+# _TTS_MAX_NEW_TOKENS_MAX (4096). Voice form max lengths are not public constants in prod;
+# values below keep tests self-contained and deterministic.
+_SPEECH_API_MAX_INSTRUCTIONS_CHARS = 500
+_SPEECH_API_MAX_NEW_TOKENS = 4096
+_VOICE_UPLOAD_MAX_CONSENT_LEN = 1024
+_VOICE_UPLOAD_MAX_REF_TEXT_CHARS = 8192
+_VOICE_UPLOAD_MAX_SPEAKER_DESCRIPTION_CHARS = 2048
+REF_AUDIO_URL = load_test_audio_data_url("qwen3_tts/clone_2.wav")
+REF_TEXT = "Okay. Yeah. I resent you. I love you. I respect you. But you know what? You blew it! And thanks to you."
+# Qwen3-TTS 0.6B expects 1024-dim speaker embeddings (see serving_speech / talker hidden_size).
+_QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM = 1024
+# Pop-only marker for ``test_speech_invalid_field_values``: omit ``ref_audio`` / ``ref_text`` so Base
+# ``speaker_embedding`` cases mirror standalone embedding-only JSON (no ref_audio clash).
+_SPEECH_INVALID_EMBEDDING_ONLY_BODY = "__speech_invalid_embedding_only_body__"
+# Batch-only: merged items inherit batch ``ref_audio`` / ``ref_text`` because ``_pick`` treats item ``None``
+# as "unset". Strip clone fields from the batch envelope before asserting CustomVoice preset errors.
+_SPEECH_BATCH_NO_CLONE_FIELDS = "__speech_batch_no_clone_fields__"
+
+_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649")
+
+
+pytestmark = [pytest.mark.slow, pytest.mark.tts]
+
+_L4_SPEECH_HW = hardware_marks(res={"cuda": "L4"})
+_SPEECH_SERVER_ARGS = ["--trust-remote-code", "--disable-log-stats"]
+
+_QWEN3_TTS_SPEECH = [
+    pytest.param(
+        OmniServerParams(
+            model="Qwen/Qwen3-TTS-12Hz-0.6B-Base",
+            stage_config_path=get_deploy_config_path("qwen3_tts.yaml"),
+            server_args=_SPEECH_SERVER_ARGS,
+        ),
+        id="qwen3_tts",
+        marks=_L4_SPEECH_HW,
+    ),
+]
+
+
+def _apply_batch_overrides(body: dict[str, object], *, loc: str, overrides: dict[str, object]) -> None:
+    if loc == "batch":
+        body.update(overrides)
+        return
+    items = body["items"]
+    assert isinstance(items, list) and len(items) >= 1
+    first = items[0]
+    assert isinstance(first, dict)
+    body["items"] = [{**first, **overrides}]
+
+
+def _pcm_wav_mono_bytes(*, duration_s: float = 1.1, sample_rate: int = 44100) -> bytes:
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wf:
+        wf.setnchannels(1)
+        wf.setsampwidth(2)
+        wf.setframerate(sample_rate)
+        nframes = max(1, int(sample_rate * duration_s))
+        wf.writeframes(b"\x00\x00" * nframes)
+    return buf.getvalue()
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST /v1/audio/speech (JSON)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_speech_malformed_json(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_audio_speech_http_request(
+        {
+            "raw_body": "{",
+            "timeout": 120,
+            "err_code": 400,
+            "err_message": ("json", "decode", "invalid"),
+        }
+    )
+
+
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_speech_missing_required_fields(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_audio_speech_http_request(
+        {
+            "json": {"model": omni_server.model},
+            "timeout": 120,
+            "err_code": 400,
+            "err_message": ("input", "field required", "missing"),
+        }
+    )
+
+
+@pytest.mark.parametrize(
+    "overrides, err_message",
+    [
+        pytest.param({"input": ""}, ("input", "empty"), id="input_empty"),
+        pytest.param({"input": "   "}, ("input", "empty"), id="input_whitespace_only"),
+        pytest.param({"voice": ""}, "Invalid voice", id="voice_empty", marks=_SKIP_ISSUE_3649),
+        pytest.param(
+            {"instructions": 123}, ("instructions", "string_type", "valid string"), id="instructions_wrong_type"
+        ),
+        pytest.param(
+            {"response_format": "mpeg"}, ("response_format", "literal_error", "wav"), id="response_format_invalid"
+        ),
+        pytest.param({"stream_format": 1}, ("stream_format", "literal_error", "audio"), id="stream_format_wrong_type"),
+        pytest.param(
+            {"stream_format": "sse"}, ("stream_format", "value_error", "audio"), id="stream_format_sse_blocked"
+        ),
+        pytest.param({"stream": "wrong_type"}, ("stream", "bool_parsing", "validation error"), id="stream_wrong_type"),
+        pytest.param(
+            {
+                "voice": "vivian",
+                "task_type": "CustomVoice",
+                "ref_audio": None,
+                "ref_text": None,
+            },
+            ("CustomVoice", "does not support"),
+            id="customvoice_task_not_supported",
+        ),
+        pytest.param(
+            {"task_type": "InvalidEnum"}, ("task_type", "literal_error", "CustomVoice"), id="task_type_invalid"
+        ),
+        pytest.param({"language": ""}, ("language", "invalid language", "chinese"), id="language_empty"),
+        pytest.param({"ref_audio": "ftp://example.com/a.wav"}, ("ref_audio", "URL"), id="ref_audio_bad_scheme"),
+        pytest.param(
+            {"ref_audio": "not_a_valid_uri"},
+            ("ref_audio", "url"),
+            id="ref_audio_invalid_uri",
+        ),
+        pytest.param({"ref_text": ["x"]}, ("ref_text", "string_type", "valid string"), id="ref_text_wrong_type"),
+        pytest.param(
+            {"x_vector_only_mode": "wrong"},
+            ("x_vector_only_mode", "valid boolean"),
+            id="x_vector_only_mode_string_not_boolean",
+        ),
+        pytest.param(
+            {"speaker_embedding": [0.01] * _QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM},
+            ("mutually exclusive", "speaker_embedding", "ref_audio"),
+            id="speaker_embedding_with_ref_audio_mutually_exclusive",
+        ),
+        pytest.param({"max_new_tokens": 0}, ("max_new_tokens", "least 1"), id="max_new_tokens_below_min"),
+        pytest.param(
+            {"max_new_tokens": _SPEECH_API_MAX_NEW_TOKENS + 1},
+            ("max_new_tokens", "exceed 4096"),
+            id="max_new_tokens_above_max",
+        ),
+        pytest.param({"seed": -1}, ("seed", "greater_than_equal", "0"), id="seed_negative"),
+        pytest.param({"seed": 2**63}, ("seed", "less_than_equal", "9223372036854775807"), id="seed_above_max"),
+        pytest.param(
+            {"initial_codec_chunk_frames": -1},
+            ("initial_codec_chunk_frames", "greater_than_equal", "0"),
+            id="initial_codec_chunk_frames_negative",
+        ),
+        pytest.param(
+            {"instructions": "x" * (_SPEECH_API_MAX_INSTRUCTIONS_CHARS + 1)},
+            ("instructions", "too long"),
+            id="instructions_exceed_api_limit",
+        ),
+        # Base clone: whitespace-only ref_text fails serving validation (not pydantic).
+        pytest.param(
+            {
+                "task_type": "Base",
+                "ref_audio": "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=",
+                "ref_text": "   ",
+            },
+            ("ref_text", "base task", "non-empty"),
+            id="ref_text_whitespace_only_base_clone",
+        ),
+        pytest.param(
+            {
+                _SPEECH_INVALID_EMBEDDING_ONLY_BODY: True,
+                "task_type": "Base",
+                "speaker_embedding": [],
+            },
+            ("speaker_embedding", "non-empty"),
+            id="base_speaker_embedding_empty",
+        ),
+        pytest.param(
+            {
+                _SPEECH_INVALID_EMBEDDING_ONLY_BODY: True,
+                "task_type": "Base",
+                "x_vector_only_mode": True,
+                "speaker_embedding": [0.01] * 512,
+            },
+            ("speaker_embedding", "dimensions", str(_QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM)),
+            id="base_speaker_embedding_wrong_dim_short",
+        ),
+        pytest.param(
+            {
+                _SPEECH_INVALID_EMBEDDING_ONLY_BODY: True,
+                "task_type": "Base",
+                "x_vector_only_mode": True,
+                "speaker_embedding": [0.01] * 2048,
+            },
+            ("speaker_embedding", "2048", "dimensions", str(_QWEN3_TTS_06B_SPEAKER_EMBEDDING_DIM)),
+            id="base_speaker_embedding_wrong_dim_long",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_speech_invalid_field_values(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    overrides: dict[str, object],
+    err_message: str | tuple[str, ...],
+) -> None:
+    ov = dict(overrides)
+    embedding_only = bool(ov.pop(_SPEECH_INVALID_EMBEDDING_ONLY_BODY, False))
+    body = {
+        "model": omni_server.model,
+        "input": "Hello.",
+        "voice": "clone",
+        "language": "English",
+        "response_format": "wav",
+        "ref_audio": REF_AUDIO_URL,
+        "ref_text": REF_TEXT,
+    }
+    if embedding_only:
+        body.pop("ref_audio", None)
+        body.pop("ref_text", None)
+    body.update(ov)
+    openai_client.send_audio_speech_http_request(
+        {"json": body, "timeout": 120, "err_code": 400, "err_message": err_message}
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST /v1/audio/speech/batch (JSON)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_speech_batch_empty_items(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_audio_speech_batch_http_request(
+        {"json": {"items": []}, "timeout": 120, "err_code": (400, 422), "err_message": ("items", "least 1 item")}
+    )
+
+
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_speech_batch_missing_items(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_audio_speech_batch_http_request(
+        {
+            "json": {"model": omni_server.model},
+            "timeout": 120,
+            "err_code": (400, 422),
+            "err_message": ("items", "field required", "missing"),
+        }
+    )
+
+
+# Align field-level checks with ``POST /v1/audio/speech`` where ``BatchSpeechRequest`` /
+# ``SpeechBatchItem`` expose the same knobs. Not on batch schema (single-speech only):
+# ``stream``, ``stream_format``, ``seed``, ``speaker_embedding`` — see dedicated speech tests.
+@pytest.mark.parametrize(
+    "loc, overrides, err_message",
+    [
+        pytest.param("item", {"input": ""}, ("input", "empty"), id="item_input_empty", marks=_SKIP_ISSUE_3649),
+        pytest.param(
+            "item",
+            {"input": "   "},
+            ("input", "empty"),
+            id="item_input_whitespace_only",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param("batch", {"voice": ""}, "Invalid voice", id="batch_voice_empty", marks=_SKIP_ISSUE_3649),
+        pytest.param("item", {"voice": ""}, "Invalid voice", id="item_voice_empty", marks=_SKIP_ISSUE_3649),
+        pytest.param(
+            "batch",
+            {
+                "voice": "nonexistent_voice_xyz",
+                "task_type": "CustomVoice",
+                "ref_audio": None,
+                "ref_text": None,
+            },
+            "Invalid voice",
+            id="batch_voice_unknown_preset",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {
+                _SPEECH_BATCH_NO_CLONE_FIELDS: True,
+                "voice": "nonexistent_voice_xyz",
+                "task_type": "CustomVoice",
+            },
+            "Invalid voice",
+            id="item_voice_unknown_preset",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "batch",
+            {"instructions": 123},
+            ("instructions", "string_type", "valid string"),
+            id="batch_instructions_wrong_type",
+        ),
+        pytest.param(
+            "item",
+            {"instructions": 123},
+            ("instructions", "string_type", "valid string"),
+            id="item_instructions_wrong_type",
+        ),
+        pytest.param(
+            "batch",
+            {"instructions": "x" * (_SPEECH_API_MAX_INSTRUCTIONS_CHARS + 1)},
+            ("instructions", "too long"),
+            id="batch_instructions_exceed_api_limit",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {"instructions": "x" * (_SPEECH_API_MAX_INSTRUCTIONS_CHARS + 1)},
+            ("instructions", "too long"),
+            id="item_instructions_exceed_api_limit",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "batch",
+            {"response_format": "mpeg"},
+            ("response_format", "literal_error", "wav"),
+            id="batch_response_format_invalid",
+        ),
+        pytest.param(
+            "item",
+            {"response_format": "mpeg"},
+            ("response_format", "literal_error", "wav"),
+            id="item_response_format_invalid",
+        ),
+        pytest.param("batch", {"speed": 0.1}, ("speed", "greater_than_equal", "0.25"), id="batch_speed_too_low"),
+        pytest.param("item", {"speed": 10.0}, ("speed", "less_than_equal", "4"), id="item_speed_too_high"),
+        pytest.param(
+            "batch",
+            {"task_type": "InvalidEnum"},
+            ("task_type", "literal_error", "CustomVoice"),
+            id="batch_task_type_invalid",
+        ),
+        pytest.param(
+            "item",
+            {"task_type": "InvalidEnum"},
+            ("task_type", "literal_error", "CustomVoice"),
+            id="item_task_type_invalid",
+        ),
+        pytest.param(
+            "batch",
+            {"language": ""},
+            ("language", "invalid language", "chinese"),
+            id="batch_language_empty",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {"language": ""},
+            ("language", "invalid language", "chinese"),
+            id="item_language_empty",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "batch",
+            {"ref_audio": "ftp://example.com/a.wav"},
+            ("ref_audio", "URL"),
+            id="batch_ref_audio_bad_scheme",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {"ref_audio": "ftp://example.com/a.wav"},
+            ("ref_audio", "URL"),
+            id="item_ref_audio_bad_scheme",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "batch",
+            {"ref_audio": "not_a_valid_uri"},
+            ("ref_audio", "url"),
+            id="batch_ref_audio_invalid_uri",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {"ref_audio": "not_a_valid_uri"},
+            ("ref_audio", "url"),
+            id="item_ref_audio_invalid_uri",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "batch",
+            {"ref_text": ["x"]},
+            ("ref_text", "string_type", "valid string"),
+            id="batch_ref_text_wrong_type",
+        ),
+        pytest.param(
+            "item",
+            {"ref_text": ["x"]},
+            ("ref_text", "string_type", "valid string"),
+            id="item_ref_text_wrong_type",
+        ),
+        pytest.param(
+            "batch",
+            {"x_vector_only_mode": "wrong_type"},
+            ("x_vector_only_mode", "bool_parsing", "validation error"),
+            id="batch_x_vector_only_mode_wrong_type",
+        ),
+        pytest.param(
+            "item",
+            {"x_vector_only_mode": "wrong_type"},
+            ("x_vector_only_mode", "bool_parsing", "validation error"),
+            id="item_x_vector_only_mode_wrong_type",
+        ),
+        pytest.param(
+            "batch",
+            {"max_new_tokens": 0},
+            ("max_new_tokens", "least 1"),
+            id="batch_max_new_tokens_below_min",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {"max_new_tokens": 0},
+            ("max_new_tokens", "least 1"),
+            id="item_max_new_tokens_below_min",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "batch",
+            {"max_new_tokens": _SPEECH_API_MAX_NEW_TOKENS + 1},
+            ("max_new_tokens", "exceed 4096"),
+            id="batch_max_new_tokens_above_max",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {"max_new_tokens": _SPEECH_API_MAX_NEW_TOKENS + 1},
+            ("max_new_tokens", "exceed 4096"),
+            id="item_max_new_tokens_above_max",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "batch",
+            {"initial_codec_chunk_frames": -1},
+            ("initial_codec_chunk_frames", "greater_than_equal", "0"),
+            id="batch_initial_codec_chunk_frames_negative",
+        ),
+        pytest.param(
+            "item",
+            {"initial_codec_chunk_frames": -1},
+            ("initial_codec_chunk_frames", "greater_than_equal", "0"),
+            id="item_initial_codec_chunk_frames_negative",
+        ),
+        # Base clone: whitespace-only ref_text (same serving check as single speech).
+        pytest.param(
+            "batch",
+            {
+                "task_type": "Base",
+                "ref_audio": "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=",
+                "ref_text": "   ",
+            },
+            ("ref_text", "base task", "non-empty"),
+            id="batch_ref_text_whitespace_only_base_clone",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "item",
+            {
+                "task_type": "Base",
+                "ref_audio": "data:audio/wav;base64,UklGRiQAAABXQVZFZm10IBAAAAABAAEARKwAAIhYAQACABAAZGF0YQAAAAA=",
+                "ref_text": "   ",
+            },
+            ("ref_text", "base task", "non-empty"),
+            id="item_ref_text_whitespace_only_base_clone",
+            marks=_SKIP_ISSUE_3649,
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_speech_batch_invalid_field_values(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    loc: str,
+    overrides: dict[str, object],
+    err_message: str | tuple[str, ...],
+) -> None:
+    ov = dict(overrides)
+    strip_clone = bool(ov.pop(_SPEECH_BATCH_NO_CLONE_FIELDS, False))
+    body = {
+        "model": omni_server.model,
+        "voice": "clone",
+        "ref_audio": REF_AUDIO_URL,
+        "ref_text": REF_TEXT,
+        "language": "English",
+        "response_format": "wav",
+        "items": [{"input": "Hello."}],
+    }
+    if strip_clone:
+        body.pop("ref_audio", None)
+        body.pop("ref_text", None)
+    _apply_batch_overrides(body, loc=loc, overrides=ov)
+    openai_client.send_audio_speech_batch_http_request(
+        {"json": body, "timeout": 120, "err_code": 400, "err_message": err_message}
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# WebSocket /v1/audio/speech/stream
+# ─────────────────────────────────────────────────────────────────────────────
+
+# ``test_speech_stream_invalid_requests``: build ``session.config`` JSON with ``omni_server.model``.
+_SPEECH_STREAM_WS_SESSION_STREAM_AUDIO_WAV_FRAMES = object()
+
+
+@pytest.mark.parametrize(
+    "send_frames_spec, err_message",
+    [
+        pytest.param("{not-json", "Invalid JSON", id="invalid_json"),
+        pytest.param(
+            json.dumps({"type": "input.text", "text": "hello"}),
+            "Expected session.config",
+            id="wrong_first_message_type",
+        ),
+        pytest.param(
+            _SPEECH_STREAM_WS_SESSION_STREAM_AUDIO_WAV_FRAMES,
+            ("invalid session config", "response_format", "pcm"),
+            id="session_config_stream_audio_requires_pcm",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_speech_stream_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    send_frames_spec: Any,
+    err_message: str | tuple[str, ...],
+) -> None:
+    """Bad first WebSocket text frames: non-JSON, wrong ``type``, or invalid ``session.config``."""
+    if send_frames_spec is _SPEECH_STREAM_WS_SESSION_STREAM_AUDIO_WAV_FRAMES:
+        send_frames = json.dumps(
+            {
+                "type": "session.config",
+                "model": omni_server.model,
+                "response_format": "wav",
+                "stream_audio": True,
+            }
+        )
+    else:
+        send_frames = send_frames_spec
+    assert isinstance(send_frames, str)
+    openai_client.send_audio_speech_stream_ws_request(
+        {
+            "send_frames": send_frames,
+            "timeout": 120,
+            "ws_max_size": None,
+            "ws_json_type": "error",
+            "err_message": err_message,
+        }
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST /v1/audio/voices (multipart + embedding-only) and DELETE /v1/audio/voices/{name}
+# ─────────────────────────────────────────────────────────────────────────────
+
+# Templates for ``test_voices_create_invalid_requests``: resolved in ``_finalize_voices_form_data``.
+_VF_LONG_CONSENT = object()
+_VF_LONG_REF_TEXT = object()
+_VF_LONG_SPEAKER_DESC = object()
+_VF_EMBEDDING_NAN_JSON = object()
+
+
+def _finalize_voices_form_data(template: dict[str, Any], uuid_hex: str) -> dict[str, Any]:
+    out: dict[str, Any] = {}
+    for key, val in template.items():
+        if val is _VF_LONG_CONSENT:
+            out[key] = "x" * (_VOICE_UPLOAD_MAX_CONSENT_LEN + 1)
+        elif val is _VF_LONG_REF_TEXT:
+            out[key] = "y" * (_VOICE_UPLOAD_MAX_REF_TEXT_CHARS + 1)
+        elif val is _VF_LONG_SPEAKER_DESC:
+            out[key] = "z" * (_VOICE_UPLOAD_MAX_SPEAKER_DESCRIPTION_CHARS + 1)
+        elif val is _VF_EMBEDDING_NAN_JSON:
+            out[key] = json.dumps([0.1] * 1023 + [float("nan")])
+        elif isinstance(val, str):
+            out[key] = val.replace("{uuid}", uuid_hex)
+        else:
+            out[key] = val
+    return out
+
+
+def _voices_upload_multipart_files(kind: str | None) -> dict[str, Any] | None:
+    """``None`` => no multipart file (embedding-only or neither)."""
+    if kind is None:
+        return None
+    if kind == "wav_ok":
+        return {"audio_sample": ("clip.wav", _pcm_wav_mono_bytes(), "audio/wav")}
+    if kind == "wav_short":
+        return {"audio_sample": ("clip.wav", _pcm_wav_mono_bytes(duration_s=0.3), "audio/wav")}
+    if kind == "wav_bad_body":
+        return {"audio_sample": ("clip.wav", b"not valid wav content", "audio/wav")}
+    if kind == "wav_pdf_type":
+        return {"audio_sample": ("clip.wav", _pcm_wav_mono_bytes(), "application/pdf")}
+    raise AssertionError(f"unknown multipart kind {kind!r}")
+
+
+@pytest.mark.parametrize(
+    "multipart_kind, form_template, err_message",
+    [
+        pytest.param(
+            None,
+            {"consent": "c1", "name": "v1"},
+            ("audio_sample", "speaker_embedding", "provided"),
+            id="neither_audio_nor_embedding",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_ok",
+            {"name": "v_miss_consent_{uuid}"},
+            ("missing", "consent", "field required"),
+            id="missing_consent",
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": "consent_ok"},
+            ("missing", "name", "field required"),
+            id="missing_name",
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": "", "name": "v_empty_consent_{uuid}"},
+            ("missing", "consent", "field required"),
+            id="empty_consent",
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": "   ", "name": "v_ws_consent_{uuid}"},
+            "consent",
+            id="whitespace_consent",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": "bad/consent", "name": "v_bad_cons_{uuid}"},
+            "consent",
+            id="consent_path_sep",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": _VF_LONG_CONSENT, "name": "v_long_cons_{uuid}"},
+            ("consent", "too long", "Failed to save"),
+            id="consent_too_long",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": "consent_ok", "name": ""},
+            ("missing", "name", "field required"),
+            id="empty_name",
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": "consent_ok", "name": "evil/name"},
+            ("voice name", "invalid voice name"),
+            id="invalid_name_path_sep",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_ok",
+            {"consent": "consent_ok", "name": "v_long_ref_{uuid}", "ref_text": _VF_LONG_REF_TEXT},
+            "ref_text",
+            id="ref_text_too_long",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_ok",
+            {
+                "consent": "consent_ok",
+                "name": "v_long_desc_{uuid}",
+                "speaker_description": _VF_LONG_SPEAKER_DESC,
+            },
+            "speaker_description",
+            id="speaker_description_too_long",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_pdf_type",
+            {"consent": "consent_ok", "name": "v_bad_mime_{uuid}"},
+            ("mime", "unsupported", "audio/mp4"),
+            id="audio_unsupported_mime",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_short",
+            {"consent": "consent_ok", "name": "v_short_{uuid}"},
+            ("too short", "reference audio"),
+            id="audio_too_short",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_bad_body",
+            {"consent": "consent_ok", "name": "v_bad_wav_{uuid}"},
+            ("decode", "audio", "Format not recognised"),
+            id="audio_decode_error",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            None,
+            {
+                "speaker_embedding": "not valid json [[[",
+                "consent": "consent_emb",
+                "name": "v_bad_json_{uuid}",
+            },
+            ("speaker_embedding", "valid JSON"),
+            id="speaker_embedding_invalid_json",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            None,
+            {
+                "speaker_embedding": json.dumps([]),
+                "consent": "consent_emb",
+                "name": "v_empty_emb_{uuid}",
+            },
+            ("speaker_embedding", "non-empty"),
+            id="speaker_embedding_empty",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            None,
+            {
+                "speaker_embedding": _VF_EMBEDDING_NAN_JSON,
+                "consent": "consent_emb",
+                "name": "v_nan_emb_{uuid}",
+            },
+            ("speaker_embedding", "finite"),
+            id="speaker_embedding_nan",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            None,
+            {
+                "speaker_embedding": json.dumps([0.01] * 512),
+                "consent": "consent_emb",
+                "name": "v_wrong_dim_{uuid}",
+            },
+            ("expected", "dimensions", "1024"),
+            id="speaker_embedding_wrong_dim",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "wav_ok",
+            {
+                "consent": "consent_emb",
+                "name": "v_both_{uuid}",
+                "speaker_embedding": json.dumps([0.01] * 1024),
+            },
+            ("mutually exclusive", "audio_sample", "speaker_embedding"),
+            id="audio_and_embedding_mutually_exclusive",
+            marks=_SKIP_ISSUE_3649,
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_voices_create_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    multipart_kind: str | None,
+    form_template: dict[str, Any],
+    err_message: str | tuple[str, ...],
+) -> None:
+    """Invalid ``POST /v1/audio/voices`` bodies (multipart or embedding-only)."""
+    uid = uuid.uuid4().hex[:8]
+    cfg: dict[str, Any] = {
+        "data": _finalize_voices_form_data(form_template, uid),
+        "timeout": 120,
+        "err_code": 400,
+        "err_message": err_message,
+    }
+    files = _voices_upload_multipart_files(multipart_kind)
+    if files is not None:
+        cfg["files"] = files
+    openai_client.send_audio_voices_create_http_request(cfg)
+
+
+@pytest.mark.parametrize(
+    "voice_name, err_code, err_message",
+    [
+        pytest.param(
+            "missing-voice-404",
+            404,
+            "not found",
+            id="not_found",
+        ),
+        pytest.param(
+            "   ",
+            404,
+            "not found",
+            id="whitespace_only",
+        ),
+        pytest.param(
+            "evil/name",
+            404,
+            "not found",
+            id="path_separator",
+        ),
+        pytest.param(
+            ".",
+            405,
+            ("Method Not Allowed", "detail"),
+            id="dot_single",
+        ),
+        pytest.param(
+            "..",
+            404,
+            "not found",
+            id="dot_dot",
+        ),
+        pytest.param(
+            "bad\x00voice",
+            404,
+            "not found",
+            id="nul_byte",
+        ),
+        pytest.param(
+            "",
+            405,
+            "Method Not Allowed",
+            id="empty_name_path",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN3_TTS_SPEECH, indirect=True)
+def test_voices_delete_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    voice_name: str,
+    err_code: int | tuple[int, ...],
+    err_message: str | tuple[str, ...],
+) -> None:
+    """DELETE ``/v1/audio/voices/{name}``: missing voice (404), odd segments (often 404), ``.``paths (405)."""
+    openai_client.send_audio_voices_delete_http_request(
+        {
+            "name": voice_name,
+            "timeout": 120,
+            "err_code": err_code,
+            "err_message": err_message,
+        }
+    )

tests/dfx/reliability/invalid_param_test/test_invalid_image_editing.py

@@ -0,0 +1,90 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""``POST /v1/images/edits`` multipart validation (live ``Qwen/Qwen-Image-Edit``)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from tests.helpers.mark import hardware_marks
+from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler
+
+pytestmark = [pytest.mark.slow, pytest.mark.diffusion]
+
+_PARAMS = [
+    pytest.param(
+        OmniServerParams(model="Qwen/Qwen-Image-Edit"),
+        id="qwen_image_edit",
+        marks=hardware_marks(res={"cuda": "H100"}),
+    ),
+]
+
+# Populated in ``test_images_edits_invalid_requests`` from ``omni_server.model``.
+_IMAGE_EDITS_SERVER_MODEL = object()
+
+
+def _finalize_edits_form_data(template: dict[str, Any], server_model: str) -> dict[str, Any]:
+    return {k: (server_model if v is _IMAGE_EDITS_SERVER_MODEL else v) for k, v in template.items()}
+
+
+@pytest.mark.parametrize(
+    "include_image, data_template, err_message, err_code",
+    [
+        pytest.param(
+            False,
+            {"prompt": "make it brighter", "model": _IMAGE_EDITS_SERVER_MODEL},
+            ("image", "required", "Unprocessable Entity"),
+            422,
+            id="missing_image",
+        ),
+        pytest.param(
+            True,
+            {"prompt": "edit", "model": "wrong-model-id"},
+            ("model", "mismatch"),
+            400,
+            id="model_mismatch",
+        ),
+        pytest.param(
+            True,
+            {"prompt": "edit", "response_format": "url"},
+            ("response_format", "b64_json", "supported"),
+            400,
+            id="unsupported_response_format",
+        ),
+        pytest.param(
+            True,
+            {"prompt": "edit", "model": _IMAGE_EDITS_SERVER_MODEL, "output_compression": "101"},
+            ("less_than_equal", "output_compression", "100"),
+            400,
+            id="output_compression_above_max",
+        ),
+        pytest.param(
+            True,
+            {"prompt": "edit", "model": _IMAGE_EDITS_SERVER_MODEL, "layers": "2"},
+            ("Invalid layers", "3", "1"),
+            400,
+            id="invalid_layers",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True)
+def test_images_edits_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    tiny_png_bytes: bytes,
+    include_image: bool,
+    data_template: dict[str, Any],
+    err_message: str | tuple[str, ...],
+    err_code: int | tuple[int, ...],
+) -> None:
+    cfg: dict[str, Any] = {
+        "data": _finalize_edits_form_data(data_template, omni_server.model),
+        "timeout": 300,
+        "err_code": err_code,
+        "err_message": err_message,
+    }
+    if include_image:
+        cfg["files"] = {"image": ("x.png", tiny_png_bytes, "image/png")}
+    openai_client.send_images_edits_http_request(cfg)

tests/dfx/reliability/invalid_param_test/test_invalid_image_generation.py

@@ -0,0 +1,142 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""``POST /v1/images/generations`` validation (live ``Qwen/Qwen-Image``)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from tests.helpers.mark import hardware_marks
+from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler
+
+pytestmark = [pytest.mark.slow, pytest.mark.diffusion]
+
+_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649")
+
+_PARAMS = [
+    pytest.param(
+        OmniServerParams(model="Qwen/Qwen-Image"),
+        id="qwen_image",
+        marks=hardware_marks(res={"cuda": "H100"}),
+    ),
+]
+
+# Full replacement body for ``test_images_generations_invalid_requests`` (no minimal prompt/model merge).
+_IMAGES_GEN_MISSING_PROMPT_JSON = object()
+
+
+def _minimal_images_gen_json(omni_server: OmniServer) -> dict[str, object]:
+    return {"prompt": "a simple red apple icon", "model": omni_server.model}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST /v1/images/generations (JSON)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "body_spec, err_message",
+    [
+        pytest.param(_IMAGES_GEN_MISSING_PROMPT_JSON, ("prompt", "Field required", "Missing"), id="missing_prompt"),
+        pytest.param(
+            {"prompt": "cat", "model": "not-the-loaded-checkpoint"},
+            ("model", "mismatch"),
+            id="model_mismatch",
+        ),
+        pytest.param({"prompt": ""}, ("prompt", "empty"), id="prompt_empty", marks=_SKIP_ISSUE_3649),
+        pytest.param({"prompt": "   "}, ("prompt", "empty"), id="prompt_whitespace_only", marks=_SKIP_ISSUE_3649),
+        pytest.param({"prompt": 123}, ("prompt", "string_type", "valid string"), id="prompt_wrong_type"),
+        pytest.param({"n": 0}, ("n", "greater_than_equal", "1"), id="n_below_min"),
+        pytest.param({"n": 11}, ("n", "less_than_equal", "10"), id="n_above_max"),
+        pytest.param({"size": "1024"}, ("size", "value_error", "WIDTHxHEIGHT"), id="size_missing_height_token"),
+        pytest.param({"size": "not-a-size"}, ("size", "value_error", "WIDTHxHEIGHT"), id="size_missing_separator"),
+        pytest.param({"size": "1024xabc"}, ("Invalid size format", "integers"), id="size_non_integer_height"),
+        pytest.param({"size": "1024x"}, ("Invalid size format", "integers"), id="size_incomplete_height"),
+        pytest.param({"size": "0x1024"}, ("Invalid size", "positive integers"), id="size_non_positive_width"),
+        pytest.param(
+            {"response_format": "url"},
+            ("response_format", "value_error", "b64_json"),
+            id="response_format_not_b64_json",
+        ),
+        pytest.param(
+            {"use_system_prompt": "bogus"},
+            ("use_system_prompt", "value_error", "dynamic"),
+            id="use_system_prompt_invalid",
+        ),
+        pytest.param(
+            {"num_inference_steps": 0},
+            ("num_inference_steps", "'greater_than_equal", "1"),
+            id="num_inference_steps_below_min",
+        ),
+        pytest.param(
+            {"num_inference_steps": 201},
+            ("num_inference_steps", "less_than_equal", "200"),
+            id="num_inference_steps_above_max",
+        ),
+        pytest.param(
+            {"guidance_scale": -1.0}, ("guidance_scale", "greater_than_equal", "0"), id="guidance_scale_negative"
+        ),
+        pytest.param(
+            {"guidance_scale": 21.0}, ("guidance_scale", "less_than_equal", "20"), id="guidance_scale_above_max"
+        ),
+        pytest.param(
+            {"true_cfg_scale": -1.0}, ("true_cfg_scale", "greater_than_equal", "0"), id="true_cfg_scale_negative"
+        ),
+        pytest.param(
+            {"true_cfg_scale": 21.0}, ("true_cfg_scale", "less_than_equal", "20"), id="true_cfg_scale_above_max"
+        ),
+        pytest.param({"layers": 2}, ("layers", "value_error", "3", "10"), id="layers_below_min"),
+        pytest.param({"layers": 11}, ("layers", "value_error", "3", "10"), id="layers_above_max"),
+        pytest.param({"seed": -1}, ("seed", "greater_than_equal", "0"), id="seed_negative", marks=_SKIP_ISSUE_3649),
+        pytest.param(
+            {"seed": 2**32},
+            ("seed", "less_than_equal", "4294967295"),
+            id="seed_above_uint32",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            {"output_format": "gif"},
+            ("output_format", "value_error", "b64_json"),
+            id="output_format_invalid",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            {"vae_use_slicing": "wrong_type"},
+            ("vae_use_slicing", "bool_parsing", "validation error"),
+            id="vae_use_slicing_wrong_type",
+        ),
+        pytest.param({"user": 123}, ("user", "string_type", "valid string"), id="user_wrong_type"),
+        pytest.param(
+            {"negative_prompt": ["noise"]},
+            ("negative_prompt", "string_type", "valid string"),
+            id="negative_prompt_wrong_type",
+        ),
+        pytest.param(
+            {"generator_device": 1},
+            ("generator_device", "string_type", "valid string"),
+            id="generator_device_wrong_type",
+        ),
+        pytest.param(
+            {"lora": {"foo": "bar"}}, ("lora", "both name and path", "required"), id="lora_missing_required_fields"
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _PARAMS, indirect=True)
+def test_images_generations_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    body_spec: Any,
+    err_message: str | tuple[str, ...],
+) -> None:
+    """Invalid JSON bodies for ``POST /v1/images/generations`` (missing prompt, model mismatch, bad fields)."""
+    if body_spec is _IMAGES_GEN_MISSING_PROMPT_JSON:
+        body: dict[str, object] = {"size": "1024x1024"}
+    else:
+        assert isinstance(body_spec, dict)
+        body = _minimal_images_gen_json(omni_server)
+        body.update(body_spec)
+    openai_client.send_images_generations_http_request(
+        {"json": body, "timeout": 300, "err_code": 400, "err_message": err_message}
+    )

tests/dfx/reliability/invalid_param_test/test_invalid_omni_chat.py

@@ -0,0 +1,216 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Invalid inputs on Qwen3-Omni: ``POST /v1/chat/completions``, ``WS /v1/video/chat/stream``, ``WS /v1/realtime``."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import pytest
+
+from tests.helpers.mark import hardware_test
+from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler
+from tests.helpers.stage_config import get_deploy_config_path
+
+pytestmark = [pytest.mark.slow, pytest.mark.omni]
+
+_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649")
+
+
+def _minimal_chat_json(omni_server: OmniServer) -> dict[str, object]:
+    """Minimal valid chat body; individual tests override one offending field."""
+    return {
+        "model": omni_server.model,
+        "messages": [{"role": "user", "content": "hello"}],
+        "stream": False,
+    }
+
+
+_QWEN3_OMNI_SERVER = [
+    pytest.param(
+        OmniServerParams(
+            model="Qwen/Qwen3-Omni-30B-A3B-Instruct",
+            stage_config_path=get_deploy_config_path("qwen3_omni_moe.yaml"),
+            server_args=["--async-chunk"],
+        ),
+        id="qwen3_omni",
+    ),
+]
+
+
+def _chat_completions_request_without_expectations(omni_server: OmniServer, case_id: str) -> dict[str, Any]:
+    """Build ``send_chat_completions_http_request`` kwargs excluding ``err_code`` / ``err_message``."""
+    if case_id == "malformed_json":
+        return {"raw_body": "{not-json", "timeout": 120}
+    if case_id == "missing_messages":
+        return {"json": {"model": omni_server.model, "stream": False}, "timeout": 120}
+    if case_id == "model_mismatch":
+        return {
+            "json": {
+                "model": "this-model-is-not-served-on-this-instance",
+                "messages": [{"role": "user", "content": "hello"}],
+                "stream": False,
+            },
+            "timeout": 120,
+        }
+    body = _minimal_chat_json(omni_server)
+    if case_id == "stream_wrong_type":
+        body["stream"] = "not-a-boolean"  # type: ignore[assignment]
+    elif case_id == "temperature_wrong_type":
+        body["temperature"] = "hot"  # type: ignore[assignment]
+    elif case_id == "max_tokens_wrong_type":
+        body["max_tokens"] = "not-an-int"  # type: ignore[assignment] # type: ignore[assignment]
+    elif case_id == "modalities_list_bad":
+        body["modalities"] = [123]  # type: ignore[assignment]
+    elif case_id == "response_format_json_schema_incomplete":
+        body["response_format"] = {"type": "json_schema"}
+    elif case_id == "logprobs_wrong_type":
+        body["logprobs"] = "yes"  # type: ignore[assignment]
+    elif case_id == "logprobs_top_without_enabled":
+        body["logprobs"] = False
+        body["top_logprobs"] = 5
+    elif case_id == "speaker_unknown":
+        body["speaker"] = "zz_invalid_qwen3_omni_chat_speaker_xyz"
+    else:
+        raise AssertionError(f"unknown chat completions invalid case_id {case_id!r}")
+    return {"json": body, "timeout": 120}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST /v1/chat/completions
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2)
+@pytest.mark.parametrize(
+    "case_id, err_code, err_message",
+    [
+        pytest.param(
+            "malformed_json",
+            400,
+            ("json_invalid", "decode error"),
+            id="malformed_json",
+        ),
+        pytest.param("missing_messages", 400, ("messages", "missing", "Field required"), id="missing_messages"),
+        pytest.param(
+            "model_mismatch",
+            404,
+            ("model", "does not exist", "NotFoundError"),
+            id="model_mismatch",
+        ),
+        pytest.param("stream_wrong_type", 400, ("stream", "bool_parsing", "not-a-boolean"), id="invalid_stream_type"),
+        pytest.param(
+            "temperature_wrong_type",
+            400,
+            ("temperature", "float_parsing", "a valid number"),
+            id="invalid_temperature_type",
+        ),
+        pytest.param(
+            "max_tokens_wrong_type", 400, ("max_tokens", "int_parsing", "integer"), id="invalid_max_tokens_type"
+        ),
+        pytest.param(
+            "modalities_list_bad",
+            400,
+            ("modalities", "value_error", ""),
+            id="modalities_list_bad_element",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "response_format_json_schema_incomplete",
+            400,
+            ("response_format", "value_error", "json_schema"),
+            id="invalid_response_format_json_schema",
+        ),
+        pytest.param("logprobs_wrong_type", 400, "logprobs", id="logprobs_wrong_type", marks=_SKIP_ISSUE_3649),
+        pytest.param(
+            "logprobs_top_without_enabled",
+            400,
+            ("top_logprobs", "logprobs", "True"),
+            id="logprobs_top_without_logprobs_true",
+        ),
+        pytest.param(
+            "speaker_unknown",
+            400,
+            ("Invalid speaker", "Supported"),
+            id="speaker_unknown_preset",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN3_OMNI_SERVER, indirect=True)
+def test_chat_completions_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    case_id: str,
+    err_code: int | tuple[int, ...],
+    err_message: str | tuple[str, ...],
+) -> None:
+    cfg = _chat_completions_request_without_expectations(omni_server, case_id)
+    cfg["err_code"] = err_code
+    cfg["err_message"] = err_message
+    openai_client.send_chat_completions_http_request(cfg)[0]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# WS /v1/video/chat/stream
+# ─────────────────────────────────────────────────────────────────────────────
+
+# First-frame JSON with ``omni_server.model`` for ``test_video_chat_stream_invalid_requests``.
+_VIDEO_CHAT_WS_SESSION_MODALITIES_SCALAR = object()
+
+
+@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2)
+@pytest.mark.parametrize(
+    "send_frames_spec, err_message",
+    [
+        pytest.param("{not-json", "Invalid JSON", id="invalid_json"),
+        pytest.param(
+            _VIDEO_CHAT_WS_SESSION_MODALITIES_SCALAR,
+            "Invalid session config",
+            id="session_modalities_scalar",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN3_OMNI_SERVER, indirect=True)
+def test_video_chat_stream_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    send_frames_spec: Any,
+    err_message: str,
+) -> None:
+    if send_frames_spec is _VIDEO_CHAT_WS_SESSION_MODALITIES_SCALAR:
+        send_frames = json.dumps({"type": "session.config", "model": omni_server.model, "modalities": "text"})
+    else:
+        send_frames = send_frames_spec
+    assert isinstance(send_frames, str)
+    openai_client.send_video_chat_stream_ws_request(
+        {
+            "send_frames": send_frames,
+            "timeout": 120,
+            "ws_max_size": None,
+            "ws_json_type": "error",
+            "err_message": err_message,
+        }
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# WS /v1/realtime
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@hardware_test(res={"cuda": "H100", "rocm": "MI325"}, num_cards=2)
+@pytest.mark.parametrize("omni_server", _QWEN3_OMNI_SERVER, indirect=True)
+def test_realtime_rejected_when_async_chunk_enabled(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+) -> None:
+    openai_client.send_realtime_ws_request(
+        {
+            "timeout": 120,
+            "ws_max_size": None,
+            "ws_json_type": "error",
+            "ws_error_code": "unsupported",
+            "err_message": "async_chunk",
+        }
+    )

tests/dfx/reliability/invalid_param_test/test_invalid_server_control.py

@@ -0,0 +1,95 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Omni sleep/wakeup JSON validation (live ``Qwen/Qwen-Image`` process)."""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+import pytest
+
+from tests.helpers.mark import hardware_marks
+from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler
+
+pytestmark = [pytest.mark.slow, pytest.mark.diffusion, pytest.mark.omni]
+
+_SKIP_ISSUE_3649 = pytest.mark.skip(reason="https://github.com/vllm-project/vllm-omni/issues/3649")
+
+_QWEN_IMAGE = [
+    pytest.param(
+        OmniServerParams(model="Qwen/Qwen-Image"),
+        id="qwen_image",
+        marks=hardware_marks(res={"cuda": "H100"}),
+    ),
+]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST omni sleep / omni wakeup (invalid JSON)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "endpoint, json_body, err_message",
+    [
+        pytest.param("sleep", {"level": 2}, ("stage_ids", "Field required", "Missing"), id="sleep_missing_stage_ids"),
+        pytest.param(
+            "sleep",
+            {"stage_ids": [], "level": 2},
+            ("stage_ids", "Field required", "Missing"),
+            id="sleep_empty_stage_ids",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param(
+            "sleep",
+            {"stage_ids": "not-a-list", "level": 2},
+            ("stage_ids", "list_type", "a valid list"),
+            id="sleep_invalid_stage_ids_type",
+        ),
+        pytest.param(
+            "sleep",
+            {"stage_ids": [0], "level": "high"},
+            ("level", "int_parsing", "a valid integer"),
+            id="sleep_invalid_level_type",
+        ),
+        pytest.param(
+            "sleep",
+            {"stage_ids": [0], "level": -1},
+            ("level", "greater"),
+            id="sleep_negative_level",
+            marks=_SKIP_ISSUE_3649,
+        ),
+        pytest.param("wakeup", {}, ("stage_ids", "missing", "Field required"), id="wakeup_missing_stage_ids"),
+        pytest.param(
+            "wakeup",
+            {"stage_ids": None},
+            ("stage_ids", "list_type", "a valid list"),
+            id="wakeup_invalid_stage_ids_type",
+        ),
+        pytest.param(
+            "wakeup",
+            {"stage_ids": []},
+            ("stage_ids", "length"),
+            id="wakeup_empty_stage_ids",
+            marks=_SKIP_ISSUE_3649,
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _QWEN_IMAGE, indirect=True)
+def test_omni_sleep_wakeup_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    endpoint: Literal["sleep", "wakeup"],
+    json_body: dict[str, Any],
+    err_message: str | tuple[str, ...],
+) -> None:
+    cfg: dict[str, Any] = {
+        "json": json_body,
+        "timeout": 120,
+        "err_code": 400,
+        "err_message": err_message,
+    }
+    if endpoint == "sleep":
+        openai_client.send_omni_sleep_http_request(cfg)
+    else:
+        openai_client.send_omni_wakeup_http_request(cfg)

tests/dfx/reliability/invalid_param_test/test_invalid_video_generation.py

@@ -0,0 +1,171 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""HTTP video job routes (live ``Wan-AI/Wan2.2-T2V-A14B-Diffusers``).
+
+WebSocket ``/v1/video/chat/stream`` and ``/v1/realtime`` invalid-input checks live in
+``test_invalid_omni_chat.py`` with the same Qwen3-Omni server module fixture (Wan cannot serve those paths).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from tests.helpers.mark import hardware_marks
+from tests.helpers.runtime import OmniServer, OmniServerParams, OpenAIClientHandler
+
+pytestmark = [pytest.mark.slow, pytest.mark.diffusion]
+
+_WAN_T2V = [
+    pytest.param(
+        OmniServerParams(model="Wan-AI/Wan2.2-T2V-A14B-Diffusers"),
+        id="wan22_t2v",
+        marks=hardware_marks(res={"cuda": "H100"}),
+    ),
+]
+
+
+def _minimal_video_form_data() -> dict[str, str]:
+    return {"prompt": "a cat walking on grass"}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST /v1/videos (async)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "form_extra, err_message",
+    [
+        pytest.param(None, ("prompt", "Field required", "Missing"), id="missing_prompt"),
+        pytest.param({"seconds": "0"}, ("seconds", "string_pattern_mismatch", "1"), id="seconds_zero"),
+        pytest.param({"size": "bad"}, ("size", "string_pattern_mismatch"), id="size_bad"),
+        pytest.param(
+            {"num_inference_steps": "0"},
+            ("num_inference_steps", "greater_than_equal", "1"),
+            id="num_inference_steps_zero",
+        ),
+        pytest.param(
+            {"guidance_scale": "25"}, ("guidance_scale", "less_than_equal", "20"), id="guidance_scale_above_max"
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True)
+def test_post_videos_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    form_extra: dict[str, str] | None,
+    err_message: str | tuple[str, ...],
+) -> None:
+    data: dict[str, str] = {} if form_extra is None else {**_minimal_video_form_data(), **form_extra}
+    openai_client.send_videos_create_http_request(
+        {"data": data, "timeout": 120, "err_code": 400, "err_message": err_message}
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# POST /v1/videos/sync
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "form_extra, err_message",
+    [
+        pytest.param(None, ("prompt", "Field required", "Missing"), id="missing_prompt"),
+        pytest.param({"seconds": "0"}, ("seconds", "string_pattern_mismatch", "1"), id="seconds_zero"),
+        pytest.param({"size": "bad"}, ("size", "string_pattern_mismatch"), id="size_bad"),
+        pytest.param(
+            {"num_inference_steps": "201"},
+            ("num_inference_steps", "less_than_equal", "200"),
+            id="num_inference_steps_above_max",
+        ),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True)
+def test_post_videos_sync_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    form_extra: dict[str, str] | None,
+    err_message: str | tuple[str, ...],
+) -> None:
+    data: dict[str, str] = {} if form_extra is None else {**_minimal_video_form_data(), **form_extra}
+    openai_client.send_videos_sync_http_request(
+        {"data": data, "timeout": 120, "err_code": 400, "err_message": err_message}
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# GET /v1/videos (list)
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "params, err_message",
+    [
+        pytest.param({"limit": -1}, ("limit", "greater_than_equal", "0"), id="invalid_limit_negative"),
+        pytest.param({"limit": 101}, ("limit", "less_than_equal", "100"), id="limit_above_max"),
+        pytest.param({"order": "newest"}, ("order", "literal_error", "asc", "desc"), id="invalid_order"),
+    ],
+)
+@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True)
+def test_get_videos_list_invalid_requests(
+    omni_server: OmniServer,
+    openai_client: OpenAIClientHandler,
+    params: dict[str, Any],
+    err_message: str | tuple[str, ...],
+) -> None:
+    openai_client.send_videos_list_http_request(
+        {"params": params, "timeout": 120, "err_code": 400, "err_message": err_message}
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# GET /v1/videos/{video_id}
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True)
+def test_get_video_invalid_requests(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_video_retrieve_http_request(
+        {
+            "video_id": "does-not-exist",
+            "timeout": 120,
+            "err_code": 404,
+            "err_message": ("not found", "video"),
+        }
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# DELETE /v1/videos/{video_id}
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True)
+def test_delete_video_invalid_requests(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_video_delete_http_request(
+        {
+            "video_id": "does-not-exist",
+            "timeout": 120,
+            "err_code": 404,
+            "err_message": ("not found", "video"),
+        }
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# GET /v1/videos/{video_id}/content
+# ─────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("omni_server", _WAN_T2V, indirect=True)
+def test_get_video_content_invalid_requests(omni_server: OmniServer, openai_client: OpenAIClientHandler) -> None:
+    openai_client.send_video_content_http_request(
+        {
+            "video_id": "does-not-exist",
+            "timeout": 120,
+            "err_code": 404,
+            "err_message": ("not found", "video"),
+        }
+    )

tests/e2e/offline_inference/test_qwen2_5_omni_autoround_w4a16_expansion.py

@@ -70,7 +70,7 @@ def test_text_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_omni_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -93,7 +93,7 @@ def test_audio_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_omni_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -114,7 +114,7 @@ def test_image_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_omni_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -135,7 +135,7 @@ def test_video_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_omni_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -162,4 +162,4 @@ def test_mix_to_audio(omni_runner, omni_runner_handler):
         "modalities": ["audio"],
     }
     response = omni_runner_handler.send_omni_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"

tests/e2e/offline_inference/test_qwen3_omni_autoround_w4a16_expansion.py

@@ -87,7 +87,7 @@ def test_text_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -111,7 +111,7 @@ def test_audio_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -133,7 +133,7 @@ def test_image_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -155,7 +155,7 @@ def test_video_to_text(omni_runner, omni_runner_handler):
         "modalities": ["text"],
     }
     response = omni_runner_handler.send_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
     assert response.text_content and len(response.text_content.strip()) > 0
 
 
@@ -177,7 +177,7 @@ def test_video_to_audio(omni_runner, omni_runner_handler):
         "modalities": ["audio"],
     }
     response = omni_runner_handler.send_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"
 
 
 # ------------------------------------------------------------------
@@ -204,4 +204,4 @@ def test_mix_to_audio(omni_runner, omni_runner_handler):
         "modalities": ["audio"],
     }
     response = omni_runner_handler.send_request(request_config)
-    assert response.success, f"Request failed: {response.error_message}"
+    assert response.success, "Request failed"

tests/e2e/online_serving/test_omnivoice_expansion.py

@@ -105,16 +105,3 @@ class TestOmniVoiceVoiceCloning:
             "min_audio_bytes": _DEFAULT_MIN_AUDIO_BYTES,
         }
         openai_client.send_audio_speech_request(request_config)
-
-    @hardware_test(res={"cuda": "L4"}, num_cards=1)
-    def test_voice_clone_invalid_ref_audio_format(self, omni_server, openai_client) -> None:
-        """Test that invalid ref_audio format returns a clear error."""
-        request_config = {
-            "model": omni_server.model,
-            "input": get_prompt("text"),
-            "ref_audio": "not_a_valid_uri",
-            "response_format": "wav",
-            "timeout": 180.0,
-            "status_code": (400, 422),
-        }
-        openai_client.send_audio_speech_request(request_config)

tests/e2e/online_serving/test_voxtral_tts.py

@@ -91,20 +91,3 @@ class TestVoxtralTTSFixedVoice:
                     "timeout": 120.0,
                 }
             )
-
-    @pytest.mark.advanced_model
-    @pytest.mark.tts
-    @hardware_test(res={"cuda": "H100"}, num_cards=1)
-    def test_speech_invalid_voice_rejected(self, omni_server, openai_client) -> None:
-        """Request with a non-existent voice should be rejected (HTTP 200 with JSON or 4xx, depending on server)."""
-        openai_client.send_audio_speech_request(
-            {
-                "model": omni_server.model,
-                "input": "This voice does not exist.",
-                "voice": "nonexistent_voice_xyz",
-                "language": "English",
-                "response_format": "wav",
-                "timeout": 120.0,
-                "err_message": "Invalid speaker",
-            }
-        )

tests/entrypoints/openai_api/test_qwen3_tts_speaker_embedding.py

@@ -5,6 +5,10 @@ E2E tests for speaker_embedding API parameter.
 
 Tests Base models (0.6B-Base and 1.7B-Base) which support voice cloning
 via pre-computed ECAPA-TDNN embeddings, bypassing ref_audio extraction.
+
+Invalid combinations (mutually exclusive with ref_audio, empty embedding,
+wrong embedding dimensions) live in
+``tests/dfx/reliability/http_invalid/test_invalid_audio_speech.py``.
 """
 
 import os
@@ -147,45 +151,6 @@ class TestSpeakerEmbeddingBase:
         assert len(response.content) > MIN_AUDIO_BYTES
         assert_not_silence(response.content)
 
-    @pytest.mark.advanced_model
-    @pytest.mark.tts
-    @hardware_test(res={"cuda": "L4"}, num_cards=1)
-    def test_speaker_embedding_mutually_exclusive_with_ref_audio(self, omni_server) -> None:
-        """Sending both speaker_embedding and ref_audio returns 400."""
-        url = f"http://{omni_server.host}:{omni_server.port}/v1/audio/speech"
-        payload = {
-            "model": omni_server.model,
-            "input": SYN_TEXT,
-            "task_type": "Base",
-            "speaker_embedding": DUMMY_EMBEDDING_1024,
-            "ref_audio": "https://example.com/audio.wav",
-            "response_format": "wav",
-        }
-        with httpx.Client(timeout=30.0) as client:
-            response = client.post(url, json=payload)
-
-        assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}"
-        assert "mutually exclusive" in response.text
-
-    @pytest.mark.advanced_model
-    @pytest.mark.tts
-    @hardware_test(res={"cuda": "L4"}, num_cards=1)
-    def test_speaker_embedding_empty_rejected(self, omni_server) -> None:
-        """Empty speaker_embedding list returns 400."""
-        url = f"http://{omni_server.host}:{omni_server.port}/v1/audio/speech"
-        payload = {
-            "model": omni_server.model,
-            "input": SYN_TEXT,
-            "task_type": "Base",
-            "speaker_embedding": [],
-            "response_format": "wav",
-        }
-        with httpx.Client(timeout=30.0) as client:
-            response = client.post(url, json=payload)
-
-        assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}"
-        assert "non-empty" in response.text
-
 
 # ── 1.7B-Base model tests (2048-dim embeddings) ──
 
@@ -216,26 +181,3 @@ class TestSpeakerEmbedding1_7B:
         assert response.headers.get("content-type") == "audio/wav"
         assert verify_wav_audio(response.content), "Response is not valid WAV"
         assert len(response.content) > MIN_AUDIO_BYTES, f"Audio too small: {len(response.content)} bytes"
-
-    @pytest.mark.advanced_model
-    @pytest.mark.tts
-    @hardware_test(res={"cuda": "L4"}, num_cards=1)
-    def test_1024_dim_on_1_7b_model_rejected(self, omni_server) -> None:
-        """1024-dim speaker_embedding on a 1.7B model (expects 2048) should fail gracefully."""
-        url = f"http://{omni_server.host}:{omni_server.port}/v1/audio/speech"
-        payload = {
-            "model": omni_server.model,
-            "input": SYN_TEXT,
-            "task_type": "Base",
-            "speaker_embedding": DUMMY_EMBEDDING_1024,
-            "x_vector_only_mode": True,
-            "response_format": "wav",
-            "max_new_tokens": MAX_NEW_TOKENS,
-        }
-        with httpx.Client(timeout=120.0) as client:
-            response = client.post(url, json=payload)
-
-        # Wrong dimensions should produce an error, not silent garbage.
-        assert response.status_code == 400, f"Expected 400, got {response.status_code}: {response.text}"
-        assert "speaker_embedding has 1024 dimensions" in response.text
-        assert "expected 2048" in response.text

tests/entrypoints/openai_api/test_serving_speech.py

@@ -1,5 +1,6 @@
 # tests/entrypoints/openai/test_serving_speech.py
 import asyncio
+import json
 import logging
 import os
 import struct
@@ -564,7 +565,6 @@ class TestSpeechAPI:
 
     def test_upload_voice_embedding_success(self, client):
         """Upload a voice via speaker_embedding JSON."""
-        import json
 
         emb = [0.1] * 1024
         data = {
@@ -585,7 +585,6 @@ class TestSpeechAPI:
 
     def test_upload_voice_embedding_appears_in_listing(self, client):
         """Embedding-uploaded voice appears in list with correct source."""
-        import json
 
         emb = [0.2] * 2048
         data = {
@@ -606,7 +605,6 @@ class TestSpeechAPI:
 
     def test_upload_voice_embedding_and_audio_mutually_exclusive(self, client):
         """Providing both audio_sample and speaker_embedding returns 400."""
-        import json
 
         emb = [0.1] * 1024
         files = {"audio_sample": ("test.wav", b"fake", "audio/wav")}
@@ -632,7 +630,6 @@ class TestSpeechAPI:
 
     def test_upload_voice_embedding_nan_rejected(self, client):
         """NaN values in speaker_embedding return 400."""
-        import json
 
         data = {
             "speaker_embedding": json.dumps([0.1] * 1023 + [float("nan")]),
@@ -930,7 +927,6 @@ class TestTTSMethods:
 
     def test_upload_voice_embedding_wrong_dims_rejected(self, speech_server):
         """Embedding uploads must match the loaded Qwen3-TTS model before being stored."""
-        import json
 
         speech_server._tts_model_type = "qwen3_tts"
         speech_server.engine_client.model_config = SimpleNamespace(

tests/helpers/assertions.py

@@ -1,12 +1,13 @@
 """Assertion and response validation helpers for tests."""
 
 import io
+import json
 import tempfile
 import threading
 import wave
 from io import BytesIO
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 if TYPE_CHECKING:
     from tests.helpers.runtime import DiffusionResponse
@@ -495,62 +496,12 @@ def assert_omni_response(response: Any, request_config: dict[str, Any], run_leve
                 )
 
 
-def _speech_combined_error_text_for_assert(response: Any) -> str:
-    """Join ``error_message`` and UTF-8 body (e.g. HTTP 200 + JSON) for substring checks."""
-    parts: list[str] = []
-    em = getattr(response, "error_message", None)
-    if em:
-        parts.append(str(em))
-    ab = getattr(response, "audio_bytes", None)
-    if ab:
-        parts.append(ab.decode("utf-8", errors="replace"))
-    return "\n".join(parts)
-
-
-def _assert_speech_error_substrings(haystack: str, err_expected: Any) -> None:
-    if isinstance(err_expected, (list, tuple)):
-        assert any(str(s) in haystack for s in err_expected), (
-            f"Expected error text to contain one of {err_expected!r}, got: {haystack!r}"
-        )
-    else:
-        assert str(err_expected) in haystack, f"Expected error text to contain {str(err_expected)!r}, got: {haystack!r}"
-
-
 def assert_audio_speech_response(response: Any, request_config: dict[str, Any], run_level: str) -> None:
-    """Validate speech API results.
+    """Validate speech API results from :class:`~tests.helpers.runtime.OmniResponse`.
 
-    If ``status_code`` is set: assert HTTP status, then optional ``err_message`` (4xx: ``error_message`` only;
-    HTTP 200: ``error_message`` and/or body, e.g. JSON with HTTP 200).
-
-    If only ``err_message`` is set: assert substring(s) on combined error text. Otherwise success/min_audio/etc.
+    Success path only. Use :meth:`OpenAIClientHandler.send_audio_speech_http_request` with ``err_code`` /
+    ``err_message`` for expected HTTP errors.
     """
-    # When present, only compare HTTP status and skip all other checks (e.g. expected 4xx).
-    # ``status_code`` may be a single int or a collection of allowed codes (e.g. ``(400, 422)``).
-    expected_status = request_config.get("status_code")
-    err_expected = request_config.get("err_message")
-    if expected_status is not None:
-        actual = getattr(response, "status_code", None)
-        assert actual is not None, (
-            "request_config has status_code but response has no .status_code "
-            "(set OmniResponse.status_code or pass a response object with status_code)"
-        )
-        if isinstance(expected_status, (list, tuple, set, frozenset)):
-            assert actual in expected_status, f"Expected HTTP status in {expected_status!r}, got {actual}"
-        else:
-            assert actual == int(expected_status), f"Expected HTTP status {int(expected_status)}, got {actual}"
-
-        if err_expected is not None:
-            if actual is not None and actual != 200:
-                msg = (getattr(response, "error_message", None) or "") + ""
-                _assert_speech_error_substrings(msg, err_expected)
-            else:
-                _assert_speech_error_substrings(_speech_combined_error_text_for_assert(response), err_expected)
-        return
-
-    if err_expected is not None:
-        _assert_speech_error_substrings(_speech_combined_error_text_for_assert(response), err_expected)
-        return
-
     assert response.success, "The request failed."
 
     # Optional floor on decoded audio size (models with very short clips may use a lower value).
@@ -599,10 +550,128 @@ def assert_diffusion_response(response: "DiffusionResponse", request_config: dic
         assert_audio_diffusion_response(response=response, request_config=request_config, run_level=run_level)
 
 
+def _http_response_body_materialize(resp: Any) -> tuple[bytes, dict[str, Any] | None]:
+    """Serialize ``HttpResponse``-like body to UTF-8 bytes and parse a JSON object when possible."""
+    jb = getattr(resp, "json_body", None)
+    if jb is not None:
+        raw = json.dumps(jb, ensure_ascii=False).encode("utf-8")
+        if isinstance(jb, dict):
+            return raw, jb
+        try:
+            parsed = json.loads(raw.decode("utf-8", errors="replace"))
+        except json.JSONDecodeError:
+            return raw, None
+        return raw, parsed if isinstance(parsed, dict) else None
+    err = getattr(resp, "error_message", None)
+    raw = (err or "").encode("utf-8", errors="replace")
+    try:
+        parsed = json.loads(raw.decode("utf-8", errors="replace"))
+    except json.JSONDecodeError:
+        return raw, None
+    return raw, parsed if isinstance(parsed, dict) else None
+
+
+def assert_err_message_in_text(
+    haystack: str,
+    err_message: str | tuple[str, ...] | list[str] | set[str] | frozenset[str],
+    *,
+    sequence_match: Literal["all", "any"] = "all",
+) -> None:
+    """Assert ``err_message`` appears in ``haystack`` (case-insensitive).
+
+    For non-string sequences: ``sequence_match='all'`` requires every substring (HTTP-style);
+    ``sequence_match='any'`` requires at least one (WebSocket first-frame JSON helpers).
+    """
+    hl = haystack.lower()
+    if isinstance(err_message, (list, tuple, set, frozenset)):
+        if sequence_match == "all":
+            missing = [s for s in err_message if str(s).lower() not in hl]
+            assert not missing, (
+                f"Expected error text to contain all of {err_message!r}; missing {missing!r}. haystack={haystack!r}"
+            )
+        else:
+            assert any(str(s).lower() in hl for s in err_message), (
+                f"Expected error text to contain one of {err_message!r}. haystack={haystack!r}"
+            )
+    else:
+        assert str(err_message).lower() in hl, f"Expected error text to contain {str(err_message)!r}, got: {haystack!r}"
+
+
+def assert_http_error(
+    resp: Any,
+    *,
+    err_code: int | tuple[int, ...] | list[int] | None = None,
+    err_message: str | tuple[str, ...] | list[str] | None = None,
+    websocket_json_message: bool = False,
+) -> dict[str, Any] | None:
+    """Validate a raw-HTTP :class:`~tests.helpers.runtime.HttpResponse`-like object.
+
+    Used by :class:`~tests.helpers.runtime.OpenAIClientHandler` ``send_*_http_request`` helpers when
+    ``request_config`` contains optional ``err_code`` and/or ``err_message``.
+
+    When ``websocket_json_message=True``, only ``json_body`` is checked (first JSON WebSocket text frame).
+    Tuple/list ``err_message`` then uses **any** substring match; HTTP mode still requires **all** pieces.
+
+    - ``err_code``: exact HTTP ``int``, or membership if a non-string sequence (e.g. ``(400, 422)``).
+      When ``err_code`` is set and the actual status is a client error in ``400..499`` other than ``404``,
+      the JSON body must include FastAPI ``detail`` and/or OpenAI-style ``error`` (2xx skips this).
+    - ``err_message``: substring match (case-insensitive) against serialized ``json_body`` and ``error_message``.
+      If ``err_message`` is a non-string sequence (``list`` / ``tuple`` / ``set`` / ``frozenset``), **every**
+      element must appear as a substring; a plain ``str`` still requires that single substring.
+    """
+    if websocket_json_message:
+        if err_code is not None:
+            raise ValueError("assert_http_error: err_code is incompatible with websocket_json_message=True")
+        jb_ws = getattr(resp, "json_body", None)
+        assert jb_ws is not None, resp
+        if err_message is None:
+            return jb_ws if isinstance(jb_ws, dict) else None
+        assert_err_message_in_text(
+            json.dumps(jb_ws, ensure_ascii=False),
+            err_message,
+            sequence_match="any",
+        )
+        return jb_ws if isinstance(jb_ws, dict) else None
+
+    if err_code is None and err_message is None:
+        return None
+
+    actual = getattr(resp, "status_code", None)
+    assert actual is not None, "response missing status_code"
+
+    if err_code is not None:
+        if isinstance(err_code, int):
+            assert actual == err_code, (resp, err_code)
+        else:
+            allowed = tuple(err_code)
+            assert actual in allowed, (resp, allowed)
+
+    body_bytes, payload = _http_response_body_materialize(resp)
+
+    if err_code is not None and actual is not None and 400 <= actual < 500 and actual != 404:
+        assert payload is not None, getattr(resp, "error_message", resp)
+        assert "detail" in payload or "error" in payload, payload
+
+    if err_message is not None:
+        pieces: list[str] = []
+        jb = getattr(resp, "json_body", None)
+        if jb is not None:
+            pieces.append(json.dumps(jb, ensure_ascii=False))
+        em = getattr(resp, "error_message", None)
+        if em:
+            pieces.append(str(em))
+        haystack = "\n".join(pieces) if pieces else body_bytes.decode("utf-8", errors="replace")
+        assert_err_message_in_text(haystack, err_message, sequence_match="all")
+
+    return payload
+
+
 __all__ = [
     "assert_audio_diffusion_response",
     "assert_audio_speech_response",
     "assert_diffusion_response",
+    "assert_err_message_in_text",
+    "assert_http_error",
     "assert_image_diffusion_response",
     "assert_image_valid",
     "assert_omni_response",

tests/helpers/runtime.py

@@ -1,5 +1,6 @@
 """Server/client/runner runtime primitives for tests."""
 
+import asyncio
 import base64
 import concurrent.futures
 import copy
@@ -19,6 +20,7 @@ from dataclasses import asdict, dataclass
 from io import BytesIO
 from pathlib import Path
 from typing import Any, NamedTuple
+from urllib.parse import quote
 
 import psutil
 import requests
@@ -37,6 +39,7 @@ from vllm.logger import init_logger
 from tests.helpers.assertions import (
     assert_audio_speech_response,
     assert_diffusion_response,
+    assert_http_error,
     assert_omni_response,
 )
 from tests.helpers.env import run_post_test_cleanup, run_pre_test_cleanup
@@ -86,6 +89,16 @@ def cleanup_dist_env_and_memory(shutdown_ray: bool = False):
             logger.warning("torch._C._host_emptyCache() only available in Pytorch >=2.5")
 
 
+def _parse_response_json(r: requests.Response) -> dict[str, Any] | list[Any] | None:
+    try:
+        data = r.json()
+        if isinstance(data, (dict, list)):
+            return data
+    except Exception:
+        pass
+    return None
+
+
 def _split_request_config_by_per_output_sizes(cfg: dict[str, Any]) -> list[dict[str, Any]] | None:
     """If ``extra_body`` has list ``height``/``width``, return one config per index (scalar h/w, ``num_outputs_per_prompt=1``)."""
     eb = cfg.get("extra_body")
@@ -647,6 +660,8 @@ class OmniServerStageCli(OmniServer):
 
 @dataclass
 class OmniResponse:
+    """Decoded multimodal / chat output from the OpenAI SDK or offline runner (not raw ``requests``)."""
+
     text_content: str | None = None
     audio_data: list[str] | None = None
     audio_content: str | None = None
@@ -656,16 +671,15 @@ class OmniResponse:
     #: OpenAI client call through response parsing and local post-process (e.g. audio decode).
     e2e_latency: float | None = None
     success: bool = False
-    error_message: str | None = None
     prompt_tokens: int | None = None
     cached_tokens: int | None = None
-    # Set for HTTP-level assertions (e.g. ``assert_audio_speech_response`` with ``status_code`` in config).
-    status_code: int | None = None
     logprobs: list | None = None
 
 
 @dataclass
 class DiffusionResponse:
+    """Decoded diffusion output from chat completions or offline runner (not raw ``requests``)."""
+
     text_content: str | None = None
     images: list[Image.Image] | None = None
     audios: list[Any] | None = None
@@ -674,7 +688,67 @@ class DiffusionResponse:
     #: ``chat.completions.create`` through local image / audio decode.
     e2e_latency: float | None = None
     success: bool = False
+
+
+@dataclass
+class HttpResponse:
+    """Normalized view of a ``requests`` response from :class:`OpenAIClientHandler` HTTP helpers."""
+
+    status_code: int
+    success: bool
     error_message: str | None = None
+    json_body: dict[str, Any] | list[Any] | None = None
+
+
+@dataclass
+class WebSocketJsonResponse:
+    """First JSON object delivered as a text WebSocket frame (streaming endpoints)."""
+
+    json_body: dict[str, Any]
+
+
+def _merge_http_expectation_kwargs(
+    base: dict[str, Any] | None,
+    *,
+    err_code: int | tuple[int, ...] | list[int] | None = None,
+    err_message: str | tuple[str, ...] | list[str] | None = None,
+) -> dict[str, Any]:
+    cfg = dict(base or {})
+    if err_code is not None:
+        cfg["err_code"] = err_code
+    if err_message is not None:
+        cfg["err_message"] = err_message
+    return cfg
+
+
+def _merge_ws_expectation_kwargs(
+    base: dict[str, Any] | None,
+    *,
+    err_message: str | tuple[str, ...] | list[str] | None = None,
+    ws_json_type: str | None = None,
+    ws_error_code: str | None = None,
+) -> dict[str, Any]:
+    cfg = dict(base or {})
+    if err_message is not None:
+        cfg["err_message"] = err_message
+    if ws_json_type is not None:
+        cfg["ws_json_type"] = ws_json_type
+    if ws_error_code is not None:
+        cfg["ws_error_code"] = ws_error_code
+    return cfg
+
+
+def _run_ws_expectations_from_request_config(cfg: dict[str, Any], resp: WebSocketJsonResponse) -> None:
+    jb = resp.json_body
+    want_type = cfg.get("ws_json_type")
+    if want_type is not None:
+        assert jb.get("type") == want_type, (jb, want_type)
+    want_code = cfg.get("ws_error_code")
+    if want_code is not None:
+        assert jb.get("code") == want_code, (jb, want_code)
+    err_message = cfg.get("err_message")
+    if err_message is not None:
+        assert_http_error(resp, err_message=err_message, websocket_json_message=True)
 
 
 def _merge_diffusion_responses(parts: list[DiffusionResponse]) -> DiffusionResponse:
@@ -739,8 +813,8 @@ class OpenAIClientHandler:
             result.e2e_latency = time.perf_counter() - wall_start
             result.success = True
         except Exception as e:
-            result.error_message = f"Stream processing error: {str(e)}"
-            print(f"Error: {result.error_message}")
+            msg = f"Stream processing error: {str(e)}"
+            print(f"Error: {msg}")
         return result
 
     def _process_non_stream_omni_response(self, chat_completion, *, wall_start: float) -> OmniResponse:
@@ -771,8 +845,8 @@ class OpenAIClientHandler:
                 result.logprobs = chat_completion.choices[0].logprobs.content
             result.success = True
         except Exception as e:
-            result.error_message = f"Non-stream processing error: {str(e)}"
-            print(f"Error: {result.error_message}")
+            msg = f"Non-stream processing error: {str(e)}"
+            print(f"Error: {msg}")
         return result
 
     def _process_diffusion_response(self, chat_completion, *, wall_start: float) -> DiffusionResponse:
@@ -811,11 +885,592 @@ class OpenAIClientHandler:
             result.e2e_latency = time.perf_counter() - wall_start
             result.success = True
         except Exception as e:
-            result.error_message = f"Diffusion response processing error: {str(e)}"
-            print(f"Error: {result.error_message}")
+            msg = f"Diffusion response processing error: {str(e)}"
+            print(f"Error: {msg}")
         return result
 
+    def _http_response_from_requests(self, r: requests.Response) -> HttpResponse:
+        payload = _parse_response_json(r)
+        ok = 200 <= r.status_code < 300
+        return HttpResponse(
+            status_code=r.status_code,
+            success=ok,
+            error_message=None if ok else (r.text[:8000] if r.text else None),
+            json_body=payload,
+        )
+
+    def send_health_http_request(
+        self,
+        request_config: dict[str, Any] | None = None,
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """GET ``/health`` (raw ``requests``).
+
+        ``request_config``: optional ``timeout`` plus optional ``err_code`` / ``err_message`` for
+        :func:`~tests.helpers.assertions.assert_http_error` (also as keyword-only args).
+        """
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = requests.get(self._build_url("/health"), timeout=float(cfg.get("timeout", 120.0)))
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_models_http_request(
+        self,
+        request_config: dict[str, Any] | None = None,
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """GET ``/v1/models``. Optional ``timeout`` and HTTP assertions (see :func:`~tests.helpers.assertions.assert_http_error`)."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = requests.get(
+            self._build_url("/v1/models"),
+            headers={"Accept": "application/json"},
+            timeout=float(cfg.get("timeout", 120.0)),
+        )
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_chat_completions_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/chat/completions`` with ``json`` or ``raw_body`` (malformed-body / contract tests)."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_json_endpoint("/v1/chat/completions", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_omni_sleep_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/omni/sleep`` — ``json`` or ``raw_body``, ``timeout``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_json_endpoint("/v1/omni/sleep", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_omni_wakeup_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/omni/wakeup``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_json_endpoint("/v1/omni/wakeup", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_audio_voices_list_http_request(
+        self,
+        request_config: dict[str, Any] | None = None,
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """GET ``/v1/audio/voices``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = requests.get(
+            self._build_url("/v1/audio/voices"),
+            headers={"Accept": "application/json"},
+            timeout=float(cfg.get("timeout", 120.0)),
+        )
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_audio_voices_create_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/audio/voices`` (multipart): ``data`` / ``files`` / ``timeout``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_form_endpoint("/v1/audio/voices", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_audio_voices_delete_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """DELETE ``/v1/audio/voices/{name}`` — requires ``name``, optional ``timeout``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        name = cfg["name"]
+        timeout = float(cfg.get("timeout", 120.0))
+        path = f"/v1/audio/voices/{quote(str(name), safe='')}"
+        r = requests.delete(
+            self._build_url(path),
+            headers={"Accept": "application/json"},
+            timeout=timeout,
+        )
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_audio_speech_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/audio/speech`` with ``json`` or ``raw_body``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_json_endpoint("/v1/audio/speech", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_audio_speech_batch_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/audio/speech/batch``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_json_endpoint("/v1/audio/speech/batch", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_audio_generate_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/audio/generate``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_json_endpoint("/v1/audio/generate", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_images_generations_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/images/generations`` — ``json`` or ``raw_body``, ``timeout``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_json_endpoint("/v1/images/generations", cfg, default_timeout=300.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_images_edits_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/images/edits`` — ``data`` / ``files`` / ``timeout``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_form_endpoint("/v1/images/edits", cfg, default_timeout=300.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_videos_create_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/videos`` (async job) — multipart ``data`` / ``files``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_form_endpoint("/v1/videos", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_videos_sync_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """POST ``/v1/videos/sync``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = self._post_form_endpoint("/v1/videos/sync", cfg, default_timeout=120.0)
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_videos_list_http_request(
+        self,
+        request_config: dict[str, Any] | None = None,
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """GET ``/v1/videos`` — optional ``params``, ``timeout``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        r = requests.get(
+            self._build_url("/v1/videos"),
+            params=cfg.get("params"),
+            headers={"Accept": "application/json"},
+            timeout=float(cfg.get("timeout", 120.0)),
+        )
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_video_retrieve_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """GET ``/v1/videos/{video_id}``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        video_id = cfg["video_id"]
+        timeout = float(cfg.get("timeout", 120.0))
+        r = requests.get(
+            self._build_url(f"/v1/videos/{quote(str(video_id), safe='')}"),
+            headers={"Accept": "application/json"},
+            timeout=timeout,
+        )
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_video_delete_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """DELETE ``/v1/videos/{video_id}``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        video_id = cfg["video_id"]
+        timeout = float(cfg.get("timeout", 120.0))
+        r = requests.delete(
+            self._build_url(f"/v1/videos/{quote(str(video_id), safe='')}"),
+            headers={"Accept": "application/json"},
+            timeout=timeout,
+        )
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def send_video_content_http_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_code: int | tuple[int, ...] | list[int] | None = None,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+    ) -> list[HttpResponse]:
+        """GET ``/v1/videos/{video_id}/content``."""
+        cfg = _merge_http_expectation_kwargs(
+            request_config,
+            err_code=err_code,
+            err_message=err_message,
+        )
+        video_id = cfg["video_id"]
+        timeout = float(cfg.get("timeout", 120.0))
+        r = requests.get(
+            self._build_url(f"/v1/videos/{quote(str(video_id), safe='')}/content"),
+            timeout=timeout,
+        )
+        resp = self._http_response_from_requests(r)
+        assert_http_error(
+            resp,
+            err_code=cfg.get("err_code"),
+            err_message=cfg.get("err_message"),
+        )
+        return [resp]
+
+    def _build_ws_url(self, path: str) -> str:
+        """Turn HTTP ``base_url`` into ``ws`` / ``wss`` for WebSocket helpers."""
+        base = self.base_url.rstrip("/")
+        suffix = "/" + path.lstrip("/")
+        if base.startswith("http://"):
+            return "ws://" + base.removeprefix("http://") + suffix
+        if base.startswith("https://"):
+            return "wss://" + base.removeprefix("https://") + suffix
+        raise ValueError(f"Unsupported base_url for WebSocket: {base!r}")
+
+    def _send_websocket_first_json_request(
+        self,
+        path: str,
+        cfg: dict[str, Any],
+    ) -> list[WebSocketJsonResponse]:
+        """Connect, optionally send text frames, return first JSON text frame as :class:`WebSocketJsonResponse`.
+
+        ``request_config`` keys:
+
+        - ``send_frames``: optional ``str`` or sequence of ``str`` raw WebSocket text frames (omit when the server
+          speaks first, e.g. ``/v1/realtime`` rejection path).
+        - ``timeout``: seconds to wait for the first inbound text frame (default ``120``).
+        - ``ws_max_size``: passed through as ``max_size`` to :func:`websockets.connect` when the key is present.
+        """
+        send_frames_raw = cfg.get("send_frames")
+        if send_frames_raw is None:
+            frames: list[str] = []
+        elif isinstance(send_frames_raw, str):
+            frames = [send_frames_raw]
+        else:
+            frames = list(send_frames_raw)
+
+        timeout = float(cfg.get("timeout", 120.0))
+        uri = self._build_ws_url(path)
+
+        connect_kw: dict[str, Any] = {}
+        if "ws_max_size" in cfg:
+            connect_kw["max_size"] = cfg["ws_max_size"]
+
+        async def _recv_first_json_object() -> WebSocketJsonResponse:
+            import websockets
+
+            async with websockets.connect(uri, **connect_kw) as ws:
+                for frame in frames:
+                    await ws.send(frame)
+                raw = await asyncio.wait_for(ws.recv(), timeout=timeout)
+                if not isinstance(raw, str):
+                    raise AssertionError(f"Expected JSON text frame from {uri}, got {type(raw).__name__}")
+                try:
+                    data = json.loads(raw)
+                except json.JSONDecodeError as exc:
+                    raise AssertionError(f"Expected JSON text frame from {uri}, body={raw[:500]!r}") from exc
+                if not isinstance(data, dict):
+                    raise AssertionError(f"Expected JSON object from {uri}, got {type(data).__name__}")
+                return WebSocketJsonResponse(json_body=data)
+
+        resp = asyncio.run(_recv_first_json_object())
+        _run_ws_expectations_from_request_config(cfg, resp)
+        return [resp]
+
+    def send_audio_speech_stream_ws_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+        ws_json_type: str | None = None,
+        ws_error_code: str | None = None,
+    ) -> list[WebSocketJsonResponse]:
+        """WebSocket ``/v1/audio/speech/stream`` — send ``send_frames`` then read first JSON text frame."""
+        cfg = _merge_ws_expectation_kwargs(
+            request_config,
+            err_message=err_message,
+            ws_json_type=ws_json_type,
+            ws_error_code=ws_error_code,
+        )
+        return self._send_websocket_first_json_request("/v1/audio/speech/stream", cfg)
+
+    def send_video_chat_stream_ws_request(
+        self,
+        request_config: dict[str, Any],
+        *,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+        ws_json_type: str | None = None,
+        ws_error_code: str | None = None,
+    ) -> list[WebSocketJsonResponse]:
+        """WebSocket ``/v1/video/chat/stream`` — send ``send_frames`` then read first JSON text frame."""
+        cfg = _merge_ws_expectation_kwargs(
+            request_config,
+            err_message=err_message,
+            ws_json_type=ws_json_type,
+            ws_error_code=ws_error_code,
+        )
+        return self._send_websocket_first_json_request("/v1/video/chat/stream", cfg)
+
+    def send_realtime_ws_request(
+        self,
+        request_config: dict[str, Any] | None = None,
+        *,
+        err_message: str | tuple[str, ...] | list[str] | None = None,
+        ws_json_type: str | None = None,
+        ws_error_code: str | None = None,
+    ) -> list[WebSocketJsonResponse]:
+        """WebSocket ``/v1/realtime`` — optional outbound frames, then first JSON text frame (often server-initiated)."""
+        cfg = _merge_ws_expectation_kwargs(
+            request_config,
+            err_message=err_message,
+            ws_json_type=ws_json_type,
+            ws_error_code=ws_error_code,
+        )
+        return self._send_websocket_first_json_request("/v1/realtime", cfg)
+
     def send_omni_request(self, request_config: dict[str, Any], request_num: int = 1) -> list[OmniResponse]:
+        """Chat completions via the OpenAI Python SDK (not raw HTTP)."""
         responses: list[OmniResponse] = []
         stream = request_config.get("stream", False)
         modalities = request_config.get("modalities", ["text", "audio"])
@@ -939,8 +1594,8 @@ class OpenAIClientHandler:
                 result.audio_format = result.audio_format.headers.get("content-type", "")
 
         except Exception as e:
-            result.error_message = f"Audio speech stream processing error: {str(e)}"
-            print(f"Error: {result.error_message}")
+            msg = f"Audio speech stream processing error: {str(e)}"
+            print(f"Error: {msg}")
 
         return result
 
@@ -978,8 +1633,8 @@ class OpenAIClientHandler:
                 result.audio_format = result.audio_format.headers.get("content-type", "")
 
         except Exception as e:
-            result.error_message = f"Audio speech non-stream processing error: {str(e)}"
-            print(f"Error: {result.error_message}")
+            msg = f"Audio speech non-stream processing error: {str(e)}"
+            print(f"Error: {msg}")
 
         return result
 
@@ -1201,6 +1856,7 @@ class OpenAIClientHandler:
         Send OpenAI requests for diffusion models.
         If ``extra_body`` has list ``height``/``width``, sends one chat completion per index in parallel
         (scalar h/w, ``num_outputs_per_prompt=1`` each) and merges images in list order.
+
         Args:
             request_config: A single request configuration dict, or a list of
                 request configuration dicts (one request per element)
@@ -1300,10 +1956,13 @@ class OpenAIClientHandler:
         self, request_config: dict[str, Any], request_num: int = 1
     ) -> list[DiffusionResponse]:
         """
-        Send native /v1/videos requests.
+        Send native /v1/videos requests: multipart ``form_data`` job create, poll until done, download content.
+
+        For raw HTTP to video routes without polling, use ``send_videos_create_http_request``, etc.
         """
         if request_num != 1:
             raise NotImplementedError("Concurrent video diffusion requests are not currently implemented")
+
         form_data = request_config.get("form_data")
         if not isinstance(form_data, dict):
             raise ValueError("Video request_config must contain 'form_data'")
@@ -1346,6 +2005,54 @@ class OpenAIClientHandler:
             self._print_client_stat("[diffusion] request#1 completed")
         return [result]
 
+    def _post_json_endpoint(
+        self,
+        path: str,
+        request_config: dict[str, Any],
+        *,
+        default_timeout: float,
+    ) -> requests.Response:
+        url = self._build_url(path)
+        timeout = float(request_config.get("timeout", default_timeout))
+        if "raw_body" in request_config:
+            raw = request_config["raw_body"]
+            payload = raw.encode("utf-8") if isinstance(raw, str) else raw
+            return requests.post(
+                url,
+                data=payload,
+                headers={"Content-Type": "application/json", "Accept": "application/json"},
+                timeout=timeout,
+            )
+        if "json" not in request_config:
+            raise ValueError(f"{path} request_config must include 'json' or 'raw_body'")
+        return requests.post(
+            url,
+            json=request_config["json"],
+            headers={"Accept": "application/json", "Content-Type": "application/json"},
+            timeout=timeout,
+        )
+
+    def _post_form_endpoint(
+        self,
+        path: str,
+        request_config: dict[str, Any],
+        *,
+        default_timeout: float = 120.0,
+    ) -> requests.Response:
+        url = self._build_url(path)
+        timeout = float(request_config.get("timeout", default_timeout))
+        data = request_config.get("data")
+        files = request_config.get("files")
+        if data is None and not files:
+            data = {}
+        return requests.post(
+            url,
+            data=data,
+            files=files,
+            headers={"Accept": "application/json"} if not files else {"Accept": "application/json"},
+            timeout=timeout,
+        )
+
     def _wait_until_video_completed(
         self, video_id: str, poll_interval_seconds: int = 2, timeout_seconds: int = 300
     ) -> None:
@@ -1693,9 +2400,9 @@ class OmniRunnerHandler:
             result.text_content = text_content
             result.success = True
         except Exception as e:
-            result.error_message = f"Output processing error: {str(e)}"
+            msg = f"Output processing error: {str(e)}"
             result.success = False
-            print(f"Error: {result.error_message}")
+            print(f"Error: {msg}")
         return result
 
     def _process_diffusion_output(self, outputs: list[OmniRequestOutput]) -> DiffusionResponse:
@@ -1831,15 +2538,11 @@ class OmniRunnerHandler:
                 mm_out = stage_out.request_output.outputs[0].multimodal_output
                 break
         if mm_out is None:
-            result = OmniResponse(success=False, error_message="No audio output from pipeline")
-            assert result.success, result.error_message
-            return result
+            raise AssertionError("No audio output from pipeline")
 
         audio_data = mm_out.get("audio")
         if audio_data is None:
-            result = OmniResponse(success=False, error_message="No audio tensor in multimodal output")
-            assert result.success, result.error_message
-            return result
+            raise AssertionError("No audio tensor in multimodal output")
 
         sr_raw = mm_out.get("sr")
         sr_val = sr_raw[-1] if isinstance(sr_raw, list) and sr_raw else sr_raw
@@ -2004,6 +2707,8 @@ def iter_omni_runner(
 
 __all__ = [
     "DiffusionResponse",
+    "HttpResponse",
+    "WebSocketJsonResponse",
     "OmniResponse",
     "OmniRunner",
     "OmniRunnerHandler",