[Test] L5 reliability test for wan2.2 and qwen3_omni model (#2972)

Signed-off-by: zhumingjue <zhumingjue@huawei.com>
Signed-off-by: [Your Name] <[Your Email]>
Signed-off-by: zhumingjue138 <zhumingjue@huawei.com>

.buildkite/pipeline.yml

@@ -1,8 +1,9 @@
 # Document 1: Buildkite loads only this block on first parse. The next step resolves docs-only skip-ci
-# from git diff, then uploads document 2. When docs-only skip applies, image-build still runs if nightly-test
-# / main NIGHTLY so upload-nightly is not skipped together with test-ready/test-merge.
-#
-# Document 2: appended after `---`; same file, read by upload_pipeline_with_skip_ci.sh (not evaluated as a second pipeline by Buildkite).
+# from git diff, then uploads document 2. When docs-only skip applies, image-build still runs for the nightly
+# exception in upload_pipeline_with_skip_ci.sh (nightly-test label or main NIGHTLY=1); otherwise upload-nightly
+# can be skipped together with test-ready/test-merge. Weekly uploads also depend on image-build.
+
+# Document 2: appended after `---`; same file, read by upload_pipeline_with_skip_ci.sh (not evaluated as a second pipeline by Buildkite). Child uploads include test-ready.yml, test-merge.yml, test-nightly.yml, and test-weekly.yml.
 steps:
   - label: ":github: Resolve skip-ci & upload pipeline"
     key: upload-ci-pipeline
@@ -61,3 +62,15 @@ steps:
       - buildkite-agent pipeline upload .buildkite/test-nightly.yml
     agents:
       queue: "cpu_queue_premerge"
+
+  # L5 Test — main+WEEKLY=1 (scheduled), or PR with weekly-test label
+  - label: "Upload Weekly Pipeline"
+    depends_on: image-build
+    key: upload-weekly-pipeline
+    if: >-
+      (build.branch == "main" && build.env("WEEKLY") == "1") ||
+      (build.branch != "main" && build.pull_request.labels includes "weekly-test")
+    commands:
+      - buildkite-agent pipeline upload .buildkite/test-weekly.yml
+    agents:
+      queue: "cpu_queue_premerge"

.buildkite/test-weekly.yml

@@ -0,0 +1,85 @@
+env:
+  VLLM_WORKER_MULTIPROC_METHOD: spawn
+  HF_HUB_DOWNLOAD_TIMEOUT: 300
+  HF_HUB_ETAG_TIMEOUT: 60
+
+steps:
+  - label: "Reliability Test - qwen3-omni"
+    timeout_in_minutes: 180
+    depends_on: upload-weekly-pipeline
+    if: build.env("WEEKLY") == "1" || build.pull_request.labels includes "weekly-test"
+    commands:
+      - pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m "slow"
+    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
+
+  - label: "Reliability Test - wan22"
+    timeout_in_minutes: 180
+    depends_on: upload-weekly-pipeline
+    if: build.env("WEEKLY") == "1" || build.pull_request.labels includes "weekly-test"
+    commands:
+      - pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m "slow"
+    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

docs/contributing/ci/CI_5levels.md

@@ -108,7 +108,8 @@ Through five levels (L1-L5) and common (Common) specifications, the system clari
         /tests/dfx/stability/tests/test_qwen3_omni.json<br>
         /tests/dfx/stability/tests/test_wan22.json<br>
         <strong>Reliability:</strong><br>
-        tests/e2e/reliability/test_{model_name}.py
+        tests/dfx/reliability/test_reliability_{model_key}.py<br>
+        (e.g. <code>test_reliability_qwen3_omni.py</code>, <code>test_reliability_wan22.py</code>)
       </td>
       <td>
         <a href="#chapter-4-l5-level-testing-stability-and-reliability-testing">Chapter 4</a><br>
@@ -576,12 +577,17 @@ L5 level testing focuses on the performance of model services under ***long-runn
 ### 4.2 Testing Content and Scope
 
 -   ***Long-term Stability (Stability) Testing***: Uses JSON under `tests/dfx/stability/tests/` (for example `test_qwen3_omni.json` and `test_wan22.json`) to run the service under moderate load for an extended period (e.g., over 12 hours), monitoring whether metrics like memory/VRAM usage, response time, and throughput degrade over time, and whether the service process remains stable.
--   ***Reliability Testing***: Uses `tests/e2e/reliability/test_{model_name}.py` to actively simulate various fault and abnormal scenarios, such as: dependent service interruption, abnormal input data, network flicker, hardware resource preemption, etc., to verify the system's fault tolerance, self-healing, and graceful degradation capabilities.
+-   ***Reliability Testing***: Uses pytest suites under `tests/dfx/reliability/` to inject controlled faults against a **live** `vllm_omni serve` instance (same **`omni_server` / `omni_server_function`** fixture style as E2E). Current suites emphasize **GPU memory pressure** (CUDA sidecar “memory hog”), **worker / runtime process kill** (`SIGKILL` on `VLLM::Worker` for Qwen3-Omni or `multiprocessing.spawn` for Wan2.2 video workers), **large multimodal chat** or **`/v1/videos`** jobs under OOM, **`/health` → 503** and **fast-fail / non-hanging concurrent** requests after kill, and **OpenAI-style 5xx error contracts** (e.g. text vs text+audio under OOM). **Post-fault recovery** checks exist where enabled (some cases may be `skip` while issues are tracked). See the Reliability `<details>` block in Section 4.4 for file-level responsibilities and CI markers (`slow`, `hardware_test`, POSIX-only kill).
 
 ### 4.3 Test Directory and Execution Files
 
 -   ***Stability Test Configuration***: `tests/dfx/stability/tests/test_qwen3_omni.json`, `tests/dfx/stability/tests/test_wan22.json` (one JSON per model / runner family)
--   ***Reliability Test Suite***: `tests/e2e/reliability/test_{model_name}.py`
+-   ***Reliability Test Suite*** (`tests/dfx/reliability/`):
+    -   `test_reliability_qwen3_omni.py` — Qwen3-Omni chat / multimodal reliability (GPU OOM, process kill, recovery, error contract under `--async-chunk` vs default).
+    -   `test_reliability_wan22.py` — Wan2.2 T2V video API reliability (`/v1/videos` under OOM and process kill, recovery).
+    -   `helpers.py` — Shared primitives used by current suites: raw HTTP probes for `/v1/chat/completions` and `/health`, OpenAI-style error parsing, GPU OOM sidecar (`inject_gpu_oom` / `stop_gpu_oom_hogs`), and `pgrep`-based process-kill injector construction (`make_process_kill_fault_injector`).
+    -   `conftest.py` — `fault_injector` and `omni_server_after_fault` / `omni_server_after_fault_function` fixtures to run a callable **after** the server is ready.
+    -   `README.md` — Short local run commands for this directory.
 
 ### 4.4 Execution Method and Example
 
@@ -662,8 +668,57 @@ All other optional parameters follow the same rules as the in Chapter 3.4.
 
 </details>
 
+<details>
+<summary> Reliability test suite (<code>tests/dfx/reliability</code>)</summary>
+
+#### Purpose and relationship to stability
+
+Reliability tests are **short fault-injection** integration runs (L5 **(b)** in `tests/dfx/reliability/README.md`). They complement **stability** JSON-driven long runs: instead of hours of steady traffic, they **perturb** the server (GPU OOM sidecar, fatal signals on selected processes) and check **failure mode** and **latency bounds** (e.g. chat or `/v1/videos` must not hang under concurrent fault-time load).
+
+#### Directory layout
+
+| Path | Responsibility |
+| ---- | -------------- |
+| `helpers.py` | Shared helpers used by current reliability suites: raw `POST`/`GET` probes (`/v1/chat/completions`, `/health`), OpenAI error parsing (`extract_openai_error_contract_from_bytes`), GPU OOM sidecar lifecycle (`inject_gpu_oom`, `stop_gpu_oom_hogs`), and process-kill injector builder (`make_process_kill_fault_injector`). |
+| `conftest.py` | Pytest fixtures: indirect `fault_injector`, `omni_server_after_fault` / `omni_server_after_fault_function` (run injector after server is ready, then yield server). |
+| `test_reliability_qwen3_omni.py` | Qwen3-Omni: OOM vs **text vs text+audio** error contract, large multimodal chat under OOM, concurrent pressure, **SIGKILL** on `VLLM::Worker`, `/health` → 503 + fast-fail + concurrent chat; optional OOM recovery scenario (may be skipped while tracked in issues). |
+| `test_reliability_wan22.py` | Wan2.2 T2V: large `/v1/videos` under OOM, **SIGKILL** on `multiprocessing.spawn` chain, health / fast-fail / concurrent video requests; optional recovery test (may be skipped). |
+| `README.md` | Minimal run / collect examples. |
+
+#### Parametrization and markers
+
+- Each test module defines a **`RELIABILITY_SCENARIOS`** list (`test_name`, `server_params`: model, `stage_config_name` or diffusion `server_args`, etc.). **`create_reliability_omni_server_params()`** in `tests/dfx/conftest.py` resolves stage paths (including XPU substitutions where applicable) and builds **`OmniServerParams`** lists consumed by **`@pytest.mark.parametrize(..., indirect=True)`** on `omni_server` or `omni_server_function`.
+- Cases are tagged **`@pytest.mark.slow`** for weekly / selective CI. GPU-heavy suites use **`@hardware_test(res={"cuda": "H100"}, num_cards=...)`** (Qwen3-Omni paths often require **2** cards; Wan2.2 video paths **1** card).
+- **Process-kill** tests use **`@pytest.mark.skipif(os.name == "nt", ...)`** because injection uses POSIX **`pgrep` / `kill`**.
+
+#### CI trigger
+
+Weekly Buildkite (`.buildkite/test-weekly.yml`) runs, for example:
+
+```bash
+pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m "slow"
+pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m "slow"
+```
+
+#### Local commands
+
+```bash
+pytest --collect-only tests/dfx/reliability
+pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m slow
+pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m slow
+```
+
+#### Adding a new model suite
+
+1. Add `test_reliability_<model>.py` under `tests/dfx/reliability/`.
+2. Define **`RELIABILITY_SCENARIOS`** and pass them through **`create_reliability_omni_server_params()`** with the correct deploy or e2e stage-config directory (same pattern as existing files).
+3. Reuse **`helpers`** for OOM / kill / raw HTTP; prefer **`assert_fault_exception()`** and **`resolve_oom_device_spec()`** from `tests/dfx/conftest.py` for consistent device selection vs stage YAML.
+4. Register **`slow`** (and **`hardware_test`** if needed); extend **`.buildkite/test-weekly.yml`** when the suite should run in weekly L5.
+
+</details>
+
 -   -   ***Stability***: `pytest -s -v tests/dfx/stability/scripts/test_stability_qwen3_omni.py` or `pytest -s -v tests/dfx/stability/scripts/test_stability_wan22.py` (or add `test_stability_<model>.py` alongside a matching JSON config)
-    -   ***Reliability***: `pytest -s -v tests/e2e/reliability/test_{model_name}.py`
+    -   ***Reliability***: `pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m slow` and/or `pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m slow` (add `test_reliability_<suite>.py` for new models)
 
 ## Summary
 

tests/dfx/conftest.py

@@ -1,13 +1,18 @@
 import json
 import os
+import re
 import subprocess
+import time
 from datetime import datetime
 from pathlib import Path
 from typing import Any
 
 import pytest
 
+from tests.dfx.reliability.helpers import list_remote_process_pids_by_pattern, post_chat_completions_raw
+from tests.helpers.runtime import OmniServerParams
 from tests.helpers.stage_config import modify_stage_config
+from vllm_omni.platforms import current_omni_platform
 
 
 def load_configs(config_path: str) -> list[dict[str, Any]]:
@@ -26,18 +31,15 @@ def load_configs(config_path: str) -> list[dict[str, Any]]:
         raise RuntimeError(f"Failed to load configuration file: {str(e)}")
 
 
-def modify_stage(default_path, updates, deletes):
-    kwargs = {}
+def modify_stage(default_path: str, updates: dict[str, Any] | None, deletes: dict[str, Any] | None) -> str:
+    kwargs: dict[str, Any] = {}
     if updates is not None:
         kwargs["updates"] = updates
     if deletes is not None:
         kwargs["deletes"] = deletes
     if kwargs:
-        path = modify_stage_config(default_path, **kwargs)
-    else:
-        path = default_path
-
-    return path
+        return modify_stage_config(default_path, **kwargs)
+    return default_path
 
 
 def _build_serve_args(serve_args: Any) -> list[str]:
@@ -111,6 +113,122 @@ def create_unique_server_params(
     return unique_params
 
 
+def configs_with_platform_stage_configs(configs: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Resolve stage config for XPU vs CUDA/ROCm."""
+    out: list[dict[str, Any]] = []
+    for config in configs:
+        config_copy = json.loads(json.dumps(config))
+        if current_omni_platform.is_xpu():
+            config_copy["server_params"]["stage_config_name"] = "xpu/qwen3_omni_ci.yaml"
+        out.append(config_copy)
+    return out
+
+
+def extract_server_args_by_test_name(configs: list[dict[str, Any]]) -> dict[str, list[str] | None]:
+    mapping: dict[str, list[str] | None] = {}
+    for cfg in configs:
+        test_name = str(cfg.get("test_name"))
+        server_params = cfg.get("server_params") or {}
+        raw_args = server_params.get("server_args")
+        mapping[test_name] = [str(item) for item in raw_args] if isinstance(raw_args, list) else None
+    return mapping
+
+
+def create_reliability_omni_server_params(
+    configs: list[dict[str, Any]], stage_configs_dir: Path
+) -> list[OmniServerParams]:
+    adjusted_configs = configs_with_platform_stage_configs(configs)
+    unique_params = create_unique_server_params(adjusted_configs, stage_configs_dir)
+    server_args_by_name = extract_server_args_by_test_name(adjusted_configs)
+    return [
+        OmniServerParams(
+            model=model,
+            stage_config_path=stage_config_path,
+            server_args=server_args_by_name.get(test_name),
+        )
+        for test_name, model, stage_config_path, _stage_overrides_json, _extra_cli_args in unique_params
+    ]
+
+
+def supports_video_generation(model_name: str) -> bool:
+    lower = model_name.lower()
+    return any(key in lower for key in ("wan", "video", "i2v", "t2v"))
+
+
+def supports_chat_generation(model_name: str) -> bool:
+    return not supports_video_generation(model_name)
+
+
+def parse_stage_devices(stage_config_path: str) -> str:
+    text = Path(stage_config_path).read_text(encoding="utf-8")
+    raw_devices: list[str] = re.findall(r"^\s*devices:\s*\"?([0-9,\s]+)\"?\s*$", text, flags=re.MULTILINE)
+    devices: set[int] = set()
+    for item in raw_devices:
+        for token in item.split(","):
+            token = token.strip()
+            if token:
+                devices.add(int(token))
+    if not devices:
+        raise ValueError(f"No runtime.devices found in stage config: {stage_config_path}")
+    return ",".join(str(x) for x in sorted(devices))
+
+
+def resolve_oom_device_spec(config: dict[str, Any], stage_config_path: str | None) -> str:
+    explicit = config.get("device")
+    if explicit is not None:
+        return str(explicit)
+    if not stage_config_path:
+        return "0"
+    return parse_stage_devices(stage_config_path)
+
+
+def assert_fault_exception(exc: Exception, error_keywords: tuple[str, ...]) -> None:
+    text = str(exc).lower()
+    assert any(key in text for key in error_keywords), f"unexpected error under fault injection: {exc}"
+
+
+def wait_chat_request_ready(host: str, port: int, model: str, timeout_sec: int = 180) -> None:
+    """Poll a minimal chat request until success."""
+    deadline = time.time() + timeout_sec
+    payload = json.dumps(
+        {
+            "model": model,
+            "messages": [{"role": "user", "content": "Say hello in one short sentence."}],
+            "stream": False,
+            "modalities": ["text"],
+        }
+    )
+    last_error: str | None = None
+    while time.time() < deadline:
+        try:
+            status, body = post_chat_completions_raw(host, port, payload)
+            if status == 200:
+                return
+            last_error = f"http={status} body={body[:200]!r}"
+        except Exception as exc:  # noqa: BLE001
+            last_error = str(exc)
+        time.sleep(2)
+    raise TimeoutError(f"runtime-teardown warmup request did not succeed within {timeout_sec}s: {last_error}")
+
+
+def assert_no_extra_worker_processes(
+    baseline_pids: set[int],
+    worker_pattern: str,
+    timeout_sec: int = 60,
+) -> None:
+    """Ensure no extra worker PIDs remain after teardown."""
+    deadline = time.time() + timeout_sec
+    while time.time() < deadline:
+        current = set(list_remote_process_pids_by_pattern(worker_pattern))
+        extra = current - baseline_pids
+        if not extra:
+            return
+        time.sleep(2)
+    current = set(list_remote_process_pids_by_pattern(worker_pattern))
+    extra = sorted(current - baseline_pids)
+    assert not extra, f"orphan worker processes remain after container teardown: {extra}"
+
+
 def create_test_parameter_mapping(configs: list[dict[str, Any]]) -> dict[str, dict]:
     mapping = {}
     for config in configs:

tests/dfx/reliability/README.md

@@ -0,0 +1,14 @@
+# L5(b) Reliability (`tests/dfx/reliability`)
+
+This directory contains reliability fault-injection tests for key models.
+The detailed RFC document is maintained outside this repository (local/internal only).
+
+- Fault helpers: `conftest.py`, `helpers.py`
+- Qwen3 tests: `test_reliability_qwen3_omni.py`
+- Wan2.2 tests: `test_reliability_wan22.py`
+
+```bash
+pytest --collect-only tests/dfx/reliability
+pytest -s -v tests/dfx/reliability/test_reliability_qwen3_omni.py -m slow
+pytest -s -v tests/dfx/reliability/test_reliability_wan22.py -m slow
+```

tests/dfx/reliability/conftest.py

@@ -0,0 +1,29 @@
+"""Pytest fixtures for reliability tests."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from tests.dfx.reliability.helpers import FaultInjector
+
+
+@pytest.fixture
+def fault_injector(request: pytest.FixtureRequest) -> FaultInjector:
+    """Indirect only: ``request.param`` must be a ``FaultInjector`` callable."""
+    return request.param
+
+
+@pytest.fixture
+def omni_server_after_fault(omni_server: Any, fault_injector: FaultInjector):
+    """After ``omni_server`` is up, run ``fault_injector(omni_server)``, then yield the server."""
+    fault_injector(omni_server)
+    yield omni_server
+
+
+@pytest.fixture
+def omni_server_after_fault_function(omni_server_function: Any, fault_injector: FaultInjector):
+    """Inject fault after function-scoped server startup, then yield server."""
+    fault_injector(omni_server_function)
+    yield omni_server_function

tests/dfx/reliability/helpers.py

@@ -0,0 +1,567 @@
+"""Shared reliability fault-injection helpers.
+
+This module keeps fault injection callable from tests directly:
+- GPU OOM (CUDA sidecar memory hog)
+- process kill by pattern and signal
+- post-ready hooks via ``fault_injector`` / ``omni_server_after_fault`` fixtures
+"""
+
+from __future__ import annotations
+
+import http.client
+import json
+import logging
+import os
+import select
+import shlex
+import signal
+import subprocess
+import sys
+import time
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from typing import Any
+
+import psutil
+import pytest
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class OomHandle:
+    """Handle for a started CUDA memory hog subprocess."""
+
+    proc: subprocess.Popen | None
+    device: int
+    target_mem_ratio: float
+    start_ts: float
+
+
+def post_chat_completions_raw(
+    host: str,
+    port: int,
+    body: bytes | str,
+    *,
+    content_type: str = "application/json",
+    timeout_sec: int = 120,
+) -> tuple[int, bytes]:
+    """POST /v1/chat/completions with raw bytes; returns (status, response_body)."""
+    conn = http.client.HTTPConnection(host, port, timeout=timeout_sec)
+    try:
+        headers = {"Content-Type": content_type}
+        payload = body.encode("utf-8") if isinstance(body, str) else body
+        conn.request("POST", "/v1/chat/completions", body=payload, headers=headers)
+        resp = conn.getresponse()
+        data = resp.read()
+        return resp.status, data
+    finally:
+        conn.close()
+
+
+def get_health_raw(host: str, port: int, *, timeout_sec: int = 20) -> tuple[int, bytes]:
+    """GET /health with stdlib HTTP client; returns (status, body)."""
+    conn = http.client.HTTPConnection(host, port, timeout=timeout_sec)
+    try:
+        conn.request("GET", "/health")
+        resp = conn.getresponse()
+        return resp.status, resp.read()
+    finally:
+        conn.close()
+
+
+def post_json_raw_http_client(
+    host: str,
+    port: int,
+    path: str,
+    payload: dict[str, Any],
+    *,
+    timeout_sec: int = 30,
+) -> tuple[int, bytes]:
+    """POST JSON to one endpoint with stdlib HTTP client; returns (status, body)."""
+    conn = http.client.HTTPConnection(host, port, timeout=timeout_sec)
+    try:
+        body = json.dumps(payload).encode("utf-8")
+        conn.request("POST", path, body=body, headers={"Content-Type": "application/json"})
+        resp = conn.getresponse()
+        return resp.status, resp.read()
+    finally:
+        conn.close()
+
+
+def post_json_raw(
+    host: str,
+    port: int,
+    path: str,
+    payload: dict[str, Any],
+    *,
+    timeout_sec: int = 30,
+) -> tuple[int, bytes]:
+    """POST JSON to one endpoint; returns (status, body)."""
+    return (
+        post_chat_completions_raw(
+            host,
+            port,
+            json.dumps(payload),
+            content_type="application/json",
+            timeout_sec=timeout_sec,
+        )
+        if path == "/v1/chat/completions"
+        else post_json_raw_http_client(
+            host,
+            port,
+            path,
+            payload,
+            timeout_sec=timeout_sec,
+        )
+    )
+
+
+def extract_openai_error_contract_from_bytes(response_body: bytes) -> dict[str, Any] | None:
+    """Best-effort parse OpenAI-style error object from raw response bytes."""
+    try:
+        payload = json.loads(response_body.decode("utf-8", errors="replace"))
+    except Exception:  # noqa: BLE001
+        return None
+    return extract_openai_error_contract_from_payload(payload)
+
+
+def extract_openai_error_contract_from_payload(payload: Any) -> dict[str, Any] | None:
+    """Best-effort parse OpenAI-style error object from decoded JSON payload."""
+    if not isinstance(payload, dict):
+        return None
+    error_obj = payload.get("error")
+    if not isinstance(error_obj, dict):
+        return None
+    if not isinstance(error_obj.get("message"), str):
+        return None
+    return error_obj
+
+
+def _build_sidecar_cmd(device: int, target_mem_ratio: float, hold_seconds: int, strict: bool) -> list[str]:
+    sidecar = r"""
+import sys
+import time
+import torch
+
+device = int(sys.argv[1])
+target_ratio = float(sys.argv[2])
+hold_seconds = int(sys.argv[3])
+strict = sys.argv[4] == "1"
+
+torch.cuda.init()
+torch.cuda.set_device(device)
+props = torch.cuda.get_device_properties(device)
+free_before, total_bytes = torch.cuda.mem_get_info(device)
+target_bytes = int(free_before * target_ratio)
+chunk_bytes = 256 * 1024 * 1024
+chunks = []
+allocated = 0
+
+while allocated < target_bytes:
+    req_bytes = min(chunk_bytes, target_bytes - allocated)
+    req_elems = max(1, req_bytes // 2)  # float16 -> 2 bytes
+    try:
+        chunk = torch.empty((req_elems,), dtype=torch.float16, device=f"cuda:{device}")
+        chunks.append(chunk)
+        allocated += chunk.numel() * 2
+    except RuntimeError:
+        break
+
+# In strict mode, keep filling with smaller chunks until allocator rejects.
+# This minimizes residual free memory and makes fault-path assertions steadier.
+if strict:
+    tail_chunk_bytes = [64 * 1024 * 1024, 16 * 1024 * 1024, 4 * 1024 * 1024, 1 * 1024 * 1024]
+    for tail_bytes in tail_chunk_bytes:
+        while True:
+            req_elems = max(1, tail_bytes // 2)
+            try:
+                chunk = torch.empty((req_elems,), dtype=torch.float16, device=f"cuda:{device}")
+                chunks.append(chunk)
+                allocated += chunk.numel() * 2
+            except RuntimeError:
+                break
+
+achieved_ratio = allocated / max(1, props.total_memory)
+achieved_free_ratio = allocated / max(1, int(free_before))
+free_after, _ = torch.cuda.mem_get_info(device)
+if strict and allocated < target_bytes:
+    print(
+        "ERROR:"
+        f"achieved_free_ratio={achieved_free_ratio:.4f};"
+        f"achieved_total_ratio={achieved_ratio:.4f};"
+        f"free_before={int(free_before)};"
+        f"free_after={int(free_after)};"
+        f"target_bytes={target_bytes};"
+        f"allocated={allocated}",
+        flush=True,
+    )
+    sys.exit(2)
+
+print(
+    "READY:"
+    f"achieved_free_ratio={achieved_free_ratio:.4f};"
+    f"achieved_total_ratio={achieved_ratio:.4f};"
+    f"free_before={int(free_before)};"
+    f"free_after={int(free_after)};"
+    f"target_bytes={target_bytes};"
+    f"allocated={allocated}",
+    flush=True,
+)
+if hold_seconds <= 0:
+    while True:
+        time.sleep(3600)
+time.sleep(hold_seconds)
+print("DONE", flush=True)
+"""
+    return [
+        sys.executable,
+        "-u",
+        "-c",
+        sidecar,
+        str(device),
+        str(target_mem_ratio),
+        str(hold_seconds),
+        "1" if strict else "0",
+    ]
+
+
+def start_gpu_oom_hog(
+    *,
+    device: int = 0,
+    target_mem_ratio: float = 0.95,
+    hold_seconds: int = 60,
+    startup_timeout_sec: int = 20,
+    strict: bool = True,
+    poll_interval_sec: float = 0.2,
+) -> OomHandle:
+    """Start a CUDA sidecar process that occupies GPU memory to trigger OOM.
+
+    Note:
+        ``target_mem_ratio`` is evaluated against free memory at injection start
+        (not total GPU memory), i.e. success gate is ``allocated / free_before``.
+        ``hold_seconds <= 0`` means keeping OOM pressure until the sidecar is
+        explicitly stopped via ``stop_gpu_oom_hog(s)``.
+    """
+    if os.name == "nt":
+        raise RuntimeError("CUDA OOM sidecar is intended for Linux CI/runtime.")
+    if not (0.0 <= target_mem_ratio < 1.0):
+        raise ValueError("target_mem_ratio should be in [0.0, 1.0).")
+
+    # Explicit opt-out for debugging: keep API shape stable while disabling injection.
+    if target_mem_ratio == 0.0:
+        print(f"[oom-sidecar][gpu={device}] DISABLED: target_mem_ratio=0.0 (no OOM injection)", flush=True)
+        return OomHandle(
+            proc=None,
+            device=device,
+            target_mem_ratio=target_mem_ratio,
+            start_ts=time.time(),
+        )
+
+    cmd = _build_sidecar_cmd(device, target_mem_ratio, hold_seconds, strict)
+    proc = subprocess.Popen(
+        cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,
+    )
+    assert proc.stdout is not None
+
+    deadline = time.time() + startup_timeout_sec
+    logs: list[str] = []
+    while time.time() < deadline:
+        ready, _, _ = select.select([proc.stdout], [], [], poll_interval_sec)
+        if ready:
+            line = proc.stdout.readline().strip()
+            if line:
+                logs.append(line)
+                print(f"[oom-sidecar][gpu={device}] {line}", flush=True)
+                if line.startswith("READY:"):
+                    return OomHandle(
+                        proc=proc,
+                        device=device,
+                        target_mem_ratio=target_mem_ratio,
+                        start_ts=time.time(),
+                    )
+                if line.startswith("ERROR:"):
+                    proc.terminate()
+                    raise RuntimeError(f"OOM sidecar failed to reach target: {line}")
+        if proc.poll() is not None:
+            break
+
+    proc.terminate()
+    if logs:
+        print(f"[oom-sidecar][gpu={device}] startup logs: {' | '.join(logs)}", flush=True)
+    raise TimeoutError(f"OOM sidecar startup timeout. logs={logs}")
+
+
+def stop_gpu_oom_hog(handle: OomHandle, *, timeout_sec: int = 5) -> None:
+    """Stop and cleanup CUDA OOM sidecar."""
+    proc = handle.proc
+    if proc is None:
+        return
+    if proc.poll() is not None:
+        return
+    proc.terminate()
+    try:
+        proc.wait(timeout=timeout_sec)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait(timeout=timeout_sec)
+
+
+def inject_gpu_oom(
+    *,
+    device: int | str | list[int] = 0,
+    target_mem_ratio: float = 0.95,
+    hold_seconds: int = 60,
+    startup_timeout_sec: int = 20,
+    strict: bool = True,
+) -> OomHandle | list[OomHandle]:
+    """Convenience wrapper to start CUDA OOM sidecar(s).
+
+    Args:
+        device: One device id (``0``), comma-separated string (``"0,1,2"``),
+            or a list of device ids (``[0, 1, 2]``).
+        hold_seconds: OOM hold time in seconds; ``<=0`` keeps pressure until
+            ``stop_gpu_oom_hogs`` is called.
+    """
+    if isinstance(device, int):
+        devices = [device]
+    elif isinstance(device, str):
+        devices = [int(x.strip()) for x in device.split(",") if x.strip()]
+    else:
+        devices = [int(x) for x in device]
+    if not devices:
+        raise ValueError("device must not be empty.")
+
+    handles = [
+        start_gpu_oom_hog(
+            device=dev,
+            target_mem_ratio=target_mem_ratio,
+            hold_seconds=hold_seconds,
+            startup_timeout_sec=startup_timeout_sec,
+            strict=strict,
+        )
+        for dev in devices
+    ]
+    if len(handles) == 1:
+        return handles[0]
+    return handles
+
+
+def stop_gpu_oom_hogs(handles: OomHandle | list[OomHandle], *, timeout_sec: int = 5) -> None:
+    """Stop one or multiple OOM sidecars."""
+    if isinstance(handles, OomHandle):
+        stop_gpu_oom_hog(handles, timeout_sec=timeout_sec)
+        return
+    for handle in handles:
+        stop_gpu_oom_hog(handle, timeout_sec=timeout_sec)
+
+
+def _runtime_teardown_ssh_target() -> str:
+    target = os.getenv("RUNTIME_TEARDOWN_SSH_TARGET", "").strip()
+    # Default to root@127.0.0.1 for same-host SSH control path.
+    return target or "root@127.0.0.1"
+
+
+def _runtime_teardown_ssh_cmd(remote_cmd: str, *, step: str | None = None) -> subprocess.CompletedProcess[str]:
+    ssh_target = _runtime_teardown_ssh_target()
+    default_reuse_opts = "-o ControlMaster=auto -o ControlPersist=10m -o ControlPath=/tmp/vllm-rt-ssh-%r@%h:%p"
+    raw_opts = os.getenv("RUNTIME_TEARDOWN_SSH_OPTS", "").strip()
+    ssh_opts = shlex.split(raw_opts or default_reuse_opts)
+    timeout_sec = int(os.getenv("RUNTIME_TEARDOWN_SSH_TIMEOUT_SEC", "600"))
+    step_prefix = f"[runtime-teardown][ssh]{f'[{step}]' if step else ''}"
+    print(f"{step_prefix} target={ssh_target} running remote command...", flush=True)
+    # IMPORTANT: SSH joins remote argv into one shell command string. If we pass
+    # ["bash", "-c", remote_cmd] as separate argv items, remote shell parsing can
+    # make `-c` consume only the first word (e.g. "docker"), causing docker help.
+    # Wrap the whole command as one quoted string for remote bash -c.
+    remote_invocation = f"bash --noprofile --norc -c {shlex.quote(remote_cmd)}"
+    try:
+        out = subprocess.run(
+            ["ssh", *ssh_opts, ssh_target, remote_invocation],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=timeout_sec,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise RuntimeError(
+            f"{step_prefix} timed out after {timeout_sec}s. Increase RUNTIME_TEARDOWN_SSH_TIMEOUT_SEC if needed."
+        ) from exc
+    print(f"{step_prefix} exit_code={out.returncode}", flush=True)
+    return out
+
+
+def list_remote_process_pids_by_pattern(pattern: str) -> list[int]:
+    """Return matched PIDs from remote host ``pgrep -f <pattern>`` via SSH."""
+    cmd = f"pgrep -f {shlex.quote(pattern)} || true"
+    out = _runtime_teardown_ssh_cmd(cmd, step="pgrep")
+    if out.returncode not in (0, 1):
+        raise RuntimeError(f"remote pgrep failed for pattern={pattern!r}: {out.stderr.strip()}")
+    return [int(item) for item in out.stdout.split() if item.strip().isdigit()]
+
+
+def inject_process_kill(
+    *,
+    grep_pattern: str,
+    signal_name: str = "SIGTERM",
+    limit: int | None = None,
+    allow_zero_match: bool = False,
+    execute_kill: bool = True,
+) -> list[int]:
+    """Kill processes matching pattern with selected signal."""
+    if os.name == "nt":
+        raise RuntimeError("process-kill helper currently supports POSIX platforms only.")
+    if not grep_pattern.strip():
+        raise ValueError("grep_pattern must not be empty.")
+
+    sig = getattr(signal, signal_name, None)
+    if sig is None:
+        raise ValueError(f"Unsupported signal_name: {signal_name}")
+
+    out = subprocess.run(
+        ["pgrep", "-f", grep_pattern],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    pids = [int(item) for item in out.stdout.split() if item.strip().isdigit()]
+    if limit is not None:
+        pids = pids[:limit]
+
+    if not pids and not allow_zero_match:
+        raise RuntimeError(f"No process matched pattern: {grep_pattern}")
+
+    if execute_kill:
+        for pid in pids:
+            os.kill(pid, sig)
+    return pids
+
+
+def _safe_proc_info(pid: int) -> tuple[str, str]:
+    """Best-effort process name/cmdline lookup for debug logging."""
+    try:
+        proc = psutil.Process(pid)
+        name = proc.name()
+        cmdline = " ".join(proc.cmdline()) or "<empty-cmdline>"
+        return name, cmdline
+    except Exception:  # noqa: BLE001
+        return "<unknown>", "<unavailable>"
+
+
+def _list_server_process_tree(server: Any) -> list[int]:
+    """Return [root, descendants...] PIDs for the current test server instance."""
+    root_proc = getattr(server, "proc", None)
+    if root_proc is None or getattr(root_proc, "pid", None) is None:
+        return []
+
+    root_pid = int(root_proc.pid)
+    try:
+        root = psutil.Process(root_pid)
+    except Exception:  # noqa: BLE001
+        return [root_pid]
+
+    descendants = [child.pid for child in root.children(recursive=True)]
+    return [root_pid, *descendants]
+
+
+def _log_server_process_tree(server: Any) -> None:
+    """Print server process tree for debugging fault injection targets."""
+    pids = _list_server_process_tree(server)
+    if not pids:
+        logger.warning("[reliability][process-kill] current server has no visible process tree")
+        return
+    for pid in pids:
+        name, cmdline = _safe_proc_info(pid)
+        print(
+            f"[reliability][process-kill] current_server_proc pid={pid} name={name} cmdline={cmdline}",
+            flush=True,
+        )
+
+
+FaultInjector = Callable[[Any], None]
+"""Callable invoked with the live ``OmniServer`` after it is ready (see ``omni_server_after_fault``)."""
+
+
+def make_process_kill_fault_injector(
+    *,
+    grep_patterns: str | Sequence[str],
+    signal_name: str = "SIGKILL",
+    limit: int = 1,
+    post_kill_wait_seconds: float = 0.0,
+) -> FaultInjector:
+    """Build a post-ready injector that kills processes matched by ``pgrep -f``.
+
+    Tries each pattern in order until at least one PID is killed. If none match,
+    the returned callable issues ``pytest.skip`` (same behavior as the previous
+    inline reliability test).
+
+    Args:
+        grep_patterns: One pattern or an ordered list of patterns.
+        signal_name: Passed to :func:`inject_process_kill` (e.g. ``SIGKILL``).
+        limit: Maximum PIDs to kill per pattern (default ``1``).
+        post_kill_wait_seconds: Optional wait time after kill before test request starts.
+    """
+    patterns: tuple[str, ...] = (grep_patterns,) if isinstance(grep_patterns, str) else tuple(grep_patterns)
+
+    def _inject(server: Any) -> None:
+        _log_server_process_tree(server)
+        server_tree = set(_list_server_process_tree(server))
+        if not server_tree:
+            logger.warning(
+                "[reliability][process-kill] no server process tree found; fallback to global pgrep matching"
+            )
+        for pattern in patterns:
+            print(
+                f"[reliability][process-kill] trying pattern={pattern} signal={signal_name} limit={limit}",
+                flush=True,
+            )
+            pids = inject_process_kill(
+                grep_pattern=pattern,
+                signal_name=signal_name,
+                limit=limit,
+                allow_zero_match=True,
+                execute_kill=False,
+            )
+            filtered = [pid for pid in pids if not server_tree or pid in server_tree]
+            if pids and not filtered:
+                logger.warning(
+                    "[reliability][process-kill] pattern=%s matched non-server pids=%s, skip them",
+                    pattern,
+                    pids,
+                )
+                continue
+            if filtered:
+                sig = getattr(signal, signal_name, None)
+                if sig is None:
+                    raise ValueError(f"Unsupported signal_name: {signal_name}")
+                for pid in filtered:
+                    name, cmdline = _safe_proc_info(pid)
+                    print(
+                        f"[reliability][process-kill] killing pid={pid} name={name} signal={signal_name} cmdline={cmdline}",
+                        flush=True,
+                    )
+                    os.kill(pid, sig)
+                print(
+                    f"[reliability][process-kill] matched pattern={pattern} killed_pids={filtered} killed_count={len(filtered)}",
+                    flush=True,
+                )
+                if post_kill_wait_seconds > 0:
+                    print(
+                        f"[reliability][process-kill] waiting {post_kill_wait_seconds:.2f}s after kill",
+                        flush=True,
+                    )
+                    time.sleep(post_kill_wait_seconds)
+                return
+        logger.warning(
+            "[reliability][process-kill] no process matched patterns=%s signal=%s limit=%s",
+            patterns,
+            signal_name,
+            limit,
+        )
+        pytest.skip("no matching runtime process found for kill injection")
+
+    return _inject

tests/dfx/reliability/test_reliability_qwen3_omni.py

@@ -0,0 +1,682 @@
+"""
+Qwen3-Omni reliability integration tests.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import errno
+import json
+import os
+import time
+from pathlib import Path
+from typing import Any, Protocol
+
+import pytest
+import torch
+
+from tests.dfx.conftest import (
+    assert_fault_exception,
+    create_reliability_omni_server_params,
+    resolve_oom_device_spec,
+)
+from tests.dfx.reliability.helpers import (
+    extract_openai_error_contract_from_bytes,
+    get_health_raw,
+    inject_gpu_oom,
+    make_process_kill_fault_injector,
+    post_chat_completions_raw,
+    post_json_raw,
+    stop_gpu_oom_hogs,
+)
+from tests.helpers.mark import hardware_test
+from tests.helpers.media import generate_synthetic_audio, generate_synthetic_image, generate_synthetic_video
+from tests.helpers.runtime import dummy_messages_from_mix_data
+
+RELIABILITY_SCENARIOS: list[dict[str, Any]] = [
+    {
+        "test_name": "qwen3_omni_reliability_async_chunk",
+        "server_params": {
+            "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
+            "stage_config_name": "qwen3_omni_moe.yaml",
+            "server_args": ["--async-chunk"],
+        },
+    },
+    {
+        "test_name": "qwen3_omni_reliability_default",
+        "server_params": {
+            "model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
+            "stage_config_name": "qwen3_omni_moe.yaml",
+            "server_args": ["--no-async-chunk"],
+        },
+    },
+]
+
+DEPLOY_CONFIGS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "vllm_omni" / "deploy"
+
+
+def _default_oom_device_spec() -> str:
+    """Use currently visible CUDA ordinals to avoid invalid device index in sidecar."""
+    count = torch.cuda.device_count()
+    if count <= 0:
+        return "0"
+    return ",".join(str(i) for i in range(count))
+
+
+OOM_INJECTION_CONFIG = {
+    "device": _default_oom_device_spec(),
+    "target_mem_ratio": 0.92,
+    "hold_seconds": 0,
+    "startup_timeout_sec": 20,
+    "strict": False,
+}
+# Post-fault recovery probe: keep the API process alive on the same GPU.
+# ``strict=True`` sidecars keep allocating until CUDA refuses, which often
+# evicts/kills the server; here we only take a fraction of *initial* free
+# memory and stop, then rely on a heavy multimodal forward to hit OOM.
+OOM_RECOVER_INJECTION_CONFIG = {
+    "device": _default_oom_device_spec(),
+    # ``strict=False`` stops at this fraction of *initial* free memory per GPU (see helpers sidecar).
+    # On multi-GPU hosts a mild ratio plus a small e2e-style mix can still leave several GB free
+    # on a data-parallel device and the request succeeds; 0.92+ pairs better with the heavy
+    # probe payload below. Raise toward 0.94 only if fault phase still never fails; avoid strict=True
+    # here so the API process is not squeezed to zero slack.
+    "target_mem_ratio": 0.8,
+    "hold_seconds": 0,
+    "startup_timeout_sec": 20,
+    "strict": False,
+}
+FAULT_ERROR_KEYWORDS = (
+    "the request failed",
+    "oom",
+    "out of memory",
+    "cuda",
+    "orchestrator",
+    "timeout",
+    "connection",
+    "500",
+    "503",
+)
+RUNTIME_WORKER_PATTERN = "VLLM::Worker"
+
+
+class _HasServeArgs(Protocol):
+    model: str
+    serve_args: list[str]
+
+
+def _get_system_prompt() -> dict:
+    return {
+        "role": "system",
+        "content": [
+            {
+                "type": "text",
+                "text": (
+                    "You are Qwen, a virtual human developed by the Qwen Team, Alibaba Group, "
+                    "capable of perceiving auditory and visual inputs, as well as generating text and speech."
+                ),
+            }
+        ],
+    }
+
+
+def _get_mix_prompt() -> str:
+    return "What is recited in the audio? What is in this image? Describe the video briefly."
+
+
+def _mix_chat_completions_probe_payload(omni_server: _HasServeArgs) -> dict[str, Any]:
+    """Heavy multimodal ``/v1/chat/completions`` body for OOM recover tests.
+
+    Uses the same synthetic sizes as ``test_reliability_fault_gpu_oom_chat_large_payload_failure``
+    (large vision tensors + long text), not the light expansion mix (224² / 5s audio): under
+    ``strict=False`` sidecars, a small mix often still completes on a GPU that retains multi-GB
+    free after the hog stops at ``target_mem_ratio``.
+    """
+    video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(1280, 720, 161)['base64']}"
+    image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+    audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(20, 1)['base64']}"
+    messages = dummy_messages_from_mix_data(
+        system_prompt=_get_system_prompt(),
+        video_data_url=video_data_url,
+        image_data_url=image_data_url,
+        audio_data_url=audio_data_url,
+        content_text="What is recited in the audio? What is in this image? What is in this video? " * 200,
+    )
+    return {
+        "model": omni_server.model,
+        "messages": messages,
+        "stream": True,
+        "modalities": ["text", "audio"],
+    }
+
+
+def _fault_keywords_match_response(*, status: int, body: bytes) -> bool:
+    """True when HTTP status or body text looks like an expected fault / OOM class outcome."""
+    if status >= 400:
+        return True
+    text = body.decode("utf-8", errors="replace").lower()
+    # Avoid matching bare ``"500"`` / ``"503"`` in successful streamed bodies (base64/SSE noise).
+    body_fault_hints = (
+        "the request failed",
+        "oom",
+        "out of memory",
+        "cuda",
+        "orchestrator",
+        "timeout",
+        "connection refused",
+        "connection reset",
+    )
+    return any(key in text for key in body_fault_hints)
+
+
+def _stage_config_path_from_omni_server(omni_server: _HasServeArgs) -> str | None:
+    args: list[str] = omni_server.serve_args
+    for i, arg in enumerate(args):
+        if arg == "--stage-configs-path" and i + 1 < len(args):
+            return args[i + 1]
+        if arg.startswith("--stage-configs-path="):
+            return arg.split("=", 1)[1]
+    return None
+
+
+def _looks_like_server_unreachable(exc: BaseException) -> bool:
+    """True when /health cannot be reached because nothing is listening (process exited)."""
+    if isinstance(exc, (ConnectionRefusedError, BrokenPipeError, ConnectionResetError)):
+        return True
+    errno_val = getattr(exc, "errno", None)
+    if isinstance(exc, OSError) and errno_val is not None:
+        return errno_val in (
+            errno.ECONNREFUSED,
+            errno.ECONNRESET,
+            errno.EPIPE,
+        )
+    msg = str(exc).lower()
+    return "connection refused" in msg or "actively refused" in msg
+
+
+QWEN_PARAMS = create_reliability_omni_server_params(RELIABILITY_SCENARIOS, DEPLOY_CONFIGS_DIR)
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_error_contract_consistent_chat_speech(
+    omni_server_function,
+) -> None:
+    """Black-box: text chat vs omni audio output should expose a consistent error contract under OOM.
+
+    Speech-style output is requested via ``/v1/chat/completions`` with ``modalities`` that include ``audio``.
+    """
+    device_spec = resolve_oom_device_spec(
+        OOM_INJECTION_CONFIG,
+        _stage_config_path_from_omni_server(omni_server_function),
+    )
+    handle = inject_gpu_oom(
+        device=device_spec,
+        target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+        hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+        startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+        strict=OOM_INJECTION_CONFIG["strict"],
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    mix_base = _mix_chat_completions_probe_payload(omni_server_function)
+    oom_contract_timeout_sec = 120
+    try:
+        chat_status, chat_body = post_json_raw(
+            host,
+            port,
+            "/v1/chat/completions",
+            {
+                "model": mix_base["model"],
+                "messages": mix_base["messages"],
+                "stream": False,
+                "modalities": ["text"],
+            },
+            timeout_sec=oom_contract_timeout_sec,
+        )
+        speech_status, speech_body = post_json_raw(
+            host,
+            port,
+            "/v1/chat/completions",
+            {
+                "model": mix_base["model"],
+                "messages": mix_base["messages"],
+                "stream": False,
+                "modalities": ["text", "audio"],
+            },
+            timeout_sec=oom_contract_timeout_sec,
+        )
+    finally:
+        stop_gpu_oom_hogs(handle)
+
+    # Under OOM pressure both chat and speech should surface runtime-class (5xx)
+    # failures instead of request-validation-class (4xx) errors.
+    chat_error = extract_openai_error_contract_from_bytes(chat_body)
+    speech_error = extract_openai_error_contract_from_bytes(speech_body)
+    print(chat_status, chat_error, speech_status, speech_error)
+
+    assert chat_status >= 500, f"expected chat error under OOM, got status={chat_status}"
+    assert speech_status >= 500, f"expected speech runtime error under OOM, got status={speech_status}"
+
+    assert chat_error is not None, f"chat error payload not OpenAI-compatible: {chat_body[:300]!r}"
+    assert speech_error is not None, f"speech error payload not OpenAI-compatible: {speech_body[:300]!r}"
+    assert "code" in chat_error, f"chat error lacks code field: {chat_error!r}"
+    assert "code" in speech_error, f"speech error lacks code field: {speech_error!r}"
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_chat_large_payload_failure(omni_server_function, openai_client_function) -> None:
+    device_spec = resolve_oom_device_spec(
+        OOM_INJECTION_CONFIG,
+        _stage_config_path_from_omni_server(omni_server_function),
+    )
+    handle = inject_gpu_oom(
+        device=device_spec,
+        target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+        hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+        startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+        strict=OOM_INJECTION_CONFIG["strict"],
+    )
+    try:
+        video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(1280, 720, 161)['base64']}"
+        image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+        audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(20, 1)['base64']}"
+        messages = dummy_messages_from_mix_data(
+            system_prompt=_get_system_prompt(),
+            video_data_url=video_data_url,
+            image_data_url=image_data_url,
+            audio_data_url=audio_data_url,
+            content_text=f"{_get_mix_prompt()} " * 200,
+        )
+        request_config = {
+            "model": omni_server_function.model,
+            "messages": messages,
+            "stream": True,
+            "key_words": {"audio": ["test"]},
+        }
+        try:
+            openai_client_function.send_omni_request(request_config, request_num=1)
+        except Exception as exc:
+            assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+        else:
+            pytest.fail("expected large chat payload request failure during GPU OOM injection")
+    finally:
+        stop_gpu_oom_hogs(handle)
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_concurrent_pressure_failure(omni_server_function, openai_client_function) -> None:
+    device_spec = resolve_oom_device_spec(
+        OOM_INJECTION_CONFIG,
+        _stage_config_path_from_omni_server(omni_server_function),
+    )
+    handle = inject_gpu_oom(
+        device=device_spec,
+        target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+        hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+        startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+        strict=OOM_INJECTION_CONFIG["strict"],
+    )
+    try:
+        video_data_url = f"data:video/mp4;base64,{generate_synthetic_video(1280, 720, 161)['base64']}"
+        image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+        audio_data_url = f"data:audio/wav;base64,{generate_synthetic_audio(20, 1)['base64']}"
+        messages = dummy_messages_from_mix_data(
+            system_prompt=_get_system_prompt(),
+            video_data_url=video_data_url,
+            image_data_url=image_data_url,
+            audio_data_url=audio_data_url,
+            content_text=f"{_get_mix_prompt()} " * 200,
+        )
+        request_config = {
+            "model": omni_server_function.model,
+            "messages": messages,
+            "stream": True,
+            "modalities": ["text", "audio"],
+            "key_words": {"audio": ["test"]},
+        }
+        try:
+            openai_client_function.send_omni_request(request_config, request_num=4)
+        except Exception as exc:
+            assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+        else:
+            pytest.fail("expected concurrent request failure under OOM injection")
+    finally:
+        stop_gpu_oom_hogs(handle)
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+    "fault_injector",
+    [
+        pytest.param(
+            make_process_kill_fault_injector(
+                grep_patterns="VLLM::Worker",
+                signal_name="SIGKILL",
+                limit=1,
+                post_kill_wait_seconds=2.0,
+            ),
+            id="runtime_process_chain",
+        ),
+    ],
+    indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_request_failure(
+    omni_server_after_fault_function, openai_client_function
+) -> None:
+    messages = dummy_messages_from_mix_data(
+        system_prompt=_get_system_prompt(),
+        content_text="What is the capital of China? Answer in 20 words.",
+    )
+    request_config = {
+        "model": omni_server_after_fault_function.model,
+        "messages": messages,
+        "stream": False,
+        "modalities": ["text"],
+        "key_words": {"text": ["beijing"]},
+    }
+    try:
+        openai_client_function.send_omni_request(request_config, request_num=1)
+    except Exception as exc:
+        assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+    else:
+        pytest.fail("expected request failure after process-kill injection")
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+    "fault_injector",
+    [
+        pytest.param(
+            make_process_kill_fault_injector(
+                grep_patterns="VLLM::Worker",
+                signal_name="SIGKILL",
+                limit=1,
+                post_kill_wait_seconds=2.0,
+            ),
+            id="runtime_process_chain",
+        ),
+    ],
+    indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_health_fast_fail_and_concurrent(
+    omni_server_after_fault_function,
+) -> None:
+    """Black-box: after worker SIGKILL, /health→503, chat fails fast, concurrent chat does not hang."""
+    host = omni_server_after_fault_function.host
+    port = omni_server_after_fault_function.port
+    model = omni_server_after_fault_function.model
+
+    deadline = time.monotonic() + 20.0
+    last_observation = ""
+    saw_503 = False
+    health_final_status: int | None = None
+    health_final_body = b""
+    while time.monotonic() < deadline:
+        try:
+            status, body = get_health_raw(host, port, timeout_sec=5)
+            last_observation = f"http={status}, body={body[:200]!r}"
+            health_final_status, health_final_body = status, body
+            if status == 503:
+                saw_503 = True
+                break
+        except Exception as exc:  # noqa: BLE001
+            last_observation = f"exception={exc!r}"
+        time.sleep(0.5)
+    assert saw_503, (
+        f"[process_kill health] expected /health to become 503 after fault injection, got {last_observation}"
+    )
+
+    payload = {
+        "model": model,
+        "messages": [{"role": "user", "content": "Say hello in one short sentence."}],
+        "stream": False,
+        "modalities": ["text"],
+    }
+    ff_status: int | None = None
+    ff_body = b""
+    ff_exc: BaseException | None = None
+    start = time.monotonic()
+    try:
+        ff_status, ff_body = post_json_raw(host, port, "/v1/chat/completions", payload, timeout_sec=20)
+        elapsed = time.monotonic() - start
+        assert elapsed < 15, f"[process_kill fast_fail] request did not fail fast after fault: {elapsed:.2f}s"
+        assert ff_status >= 500, (
+            f"[process_kill fast_fail] expected server-side failure after fault, "
+            f"got status={ff_status}, body={ff_body[:200]!r}"
+        )
+    except Exception as exc:
+        ff_exc = exc
+        elapsed = time.monotonic() - start
+        assert elapsed < 15, f"[process_kill fast_fail] request exception was too slow after fault: {elapsed:.2f}s"
+
+    payload_json = json.dumps(
+        {
+            "model": model,
+            "messages": [{"role": "user", "content": "What is the capital of China? Answer in one word."}],
+            "stream": False,
+            "modalities": ["text"],
+        }
+    )
+    start = time.monotonic()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
+        futures = [
+            executor.submit(
+                post_chat_completions_raw,
+                host,
+                port,
+                payload_json,
+                timeout_sec=20,
+            )
+            for _ in range(4)
+        ]
+        done, pending = concurrent.futures.wait(
+            futures,
+            timeout=30,
+            return_when=concurrent.futures.ALL_COMPLETED,
+        )
+
+    elapsed = time.monotonic() - start
+    assert not pending, f"[process_kill concurrent] some fault-time requests hung: pending={len(pending)}"
+    assert elapsed < 30, f"[process_kill concurrent] fault-time request convergence is too slow: {elapsed:.2f}s"
+
+    fault_observed = False
+    conc_debug: list[Any] = []
+    for future in done:
+        try:
+            status, body = future.result()
+            conc_debug.append((status, body[:200]))
+            if status >= 500:
+                fault_observed = True
+        except Exception as exc:
+            conc_debug.append(repr(exc))
+            fault_observed = True
+    # DEBUG: remove before merge
+    print(
+        health_final_status,
+        health_final_body[:200],
+        ff_status,
+        ff_body[:200],
+        ff_exc,
+        conc_debug,
+    )
+    assert fault_observed, (
+        "[process_kill concurrent] expected at least one request to fail after process-kill fault injection"
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.skip(reason="issue#2327")
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_state_converges_after_fault_removed(
+    omni_server_function,
+) -> None:
+    """Black-box: under bounded GPU pressure, a heavy mix chat fails; after hog stops, the same request succeeds.
+
+    Uses a milder sidecar than ``OOM_INJECTION_CONFIG`` (``strict=False`` and a capped
+    ``target_mem_ratio``) so the serving process stays alive: pressure should starve
+    the multimodal forward pass, not exhaust the whole device allocator.
+
+    Fault and recovery phases use raw ``/v1/chat/completions`` HTTP so tests can assert
+    on status codes and JSON error payloads (or fault keywords in streamed bodies).
+    """
+    device_spec = resolve_oom_device_spec(
+        OOM_RECOVER_INJECTION_CONFIG,
+        _stage_config_path_from_omni_server(omni_server_function),
+    )
+    handle = inject_gpu_oom(
+        device=device_spec,
+        target_mem_ratio=OOM_RECOVER_INJECTION_CONFIG["target_mem_ratio"],
+        hold_seconds=OOM_RECOVER_INJECTION_CONFIG["hold_seconds"],
+        startup_timeout_sec=OOM_RECOVER_INJECTION_CONFIG["startup_timeout_sec"],
+        strict=OOM_RECOVER_INJECTION_CONFIG["strict"],
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    mix_payload = _mix_chat_completions_probe_payload(omni_server_function)
+    mix_chat_timeout_sec = 180
+    print(
+        "[oom-recover] injection ready "
+        f"device={device_spec} target_mem_ratio={OOM_RECOVER_INJECTION_CONFIG['target_mem_ratio']} "
+        f"strict={OOM_RECOVER_INJECTION_CONFIG['strict']} hold_seconds={OOM_RECOVER_INJECTION_CONFIG['hold_seconds']}"
+    )
+    print(f"[oom-recover] target endpoint http://{host}:{port}/v1/chat/completions")
+
+    failure_observed = False
+    fault_status: int | None = None
+    fault_body = b""
+    try:
+        try:
+            print(f"[oom-recover] fault-phase request start timeout={mix_chat_timeout_sec}s")
+            fault_status, fault_body = post_json_raw(
+                host,
+                port,
+                "/v1/chat/completions",
+                mix_payload,
+                timeout_sec=mix_chat_timeout_sec,
+            )
+            print(f"[oom-recover] fault-phase request done status={fault_status} body_prefix={fault_body[:200]!r}")
+        except Exception as exc:
+            failure_observed = True
+            print(f"[oom-recover] fault-phase request raised {type(exc).__name__}: {exc!r}")
+            assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+        else:
+            assert fault_status is not None
+            failure_observed = fault_status != 200 or _fault_keywords_match_response(
+                status=fault_status,
+                body=fault_body,
+            )
+            assert failure_observed, (
+                "expected mix multimodal request failure while OOM pressure is active; "
+                f"status={fault_status}, body_prefix={fault_body[:500]!r}"
+            )
+            if fault_status >= 400:
+                err = extract_openai_error_contract_from_bytes(fault_body)
+                if err is None:
+                    assert _fault_keywords_match_response(status=fault_status, body=fault_body), (
+                        "non-2xx without OpenAI-style error object should still mention fault hints in body; "
+                        f"status={fault_status}, body_prefix={fault_body[:500]!r}"
+                    )
+                else:
+                    assert isinstance(err.get("message"), str) and err["message"].strip(), (
+                        f"structured error should carry a non-empty message: {err!r}"
+                    )
+            else:
+                assert _fault_keywords_match_response(status=fault_status, body=fault_body), (
+                    "HTTP 200 under pressure should still surface fault hints in the response body; "
+                    f"body_prefix={fault_body[:500]!r}"
+                )
+    finally:
+        print("[oom-recover] stopping OOM sidecar(s)")
+        stop_gpu_oom_hogs(handle)
+        print("[oom-recover] OOM sidecar(s) stopped")
+
+    recovery_deadline = time.monotonic() + 90.0
+    terminal_health: int | None = None
+    unreachable_streak = 0
+    last_health_exc: BaseException | None = None
+    while time.monotonic() < recovery_deadline:
+        try:
+            status, _ = get_health_raw(host, port, timeout_sec=5)
+            unreachable_streak = 0
+            if status in (200, 503):
+                terminal_health = status
+                print(f"[oom-recover] terminal health reached status={terminal_health}")
+                break
+        except Exception as exc:
+            last_health_exc = exc
+            if _looks_like_server_unreachable(exc):
+                unreachable_streak += 1
+                if unreachable_streak >= 5:
+                    pytest.fail(
+                        "after OOM sidecar stopped, /health is unreachable (connection refused / reset). "
+                        "The APIServer process likely exited (e.g. orchestrator thread crash under OOM); "
+                        "this test expects the server to stay up for post-fault health polling. "
+                        f"last_exc={last_health_exc!r}"
+                    )
+            else:
+                unreachable_streak = 0
+        time.sleep(1.0)
+    else:
+        pytest.fail(
+            "server did not converge to a terminal health state after OOM pressure was removed; "
+            f"last_health_exc={last_health_exc!r}"
+        )
+
+    probe_payload = {
+        "model": omni_server_function.model,
+        "messages": [{"role": "user", "content": "What is the capital of China? Answer in one word."}],
+        "stream": False,
+        "modalities": ["text"],
+    }
+    start = time.monotonic()
+    post_mix_status: int | None = None
+    post_mix_body = b""
+    request_status: int | None = None
+    try:
+        assert terminal_health is not None
+        if terminal_health == 200:
+            print(f"[oom-recover] recovery-phase mix request start timeout={mix_chat_timeout_sec}s")
+            post_mix_status, post_mix_body = post_json_raw(
+                host,
+                port,
+                "/v1/chat/completions",
+                mix_payload,
+                timeout_sec=mix_chat_timeout_sec,
+            )
+            print(
+                "[oom-recover] recovery-phase mix request done "
+                f"status={post_mix_status} body_prefix={post_mix_body[:200]!r}"
+            )
+        else:
+            print("[oom-recover] terminal health=503, send lightweight probe request")
+            request_status, _ = post_json_raw(host, port, "/v1/chat/completions", probe_payload, timeout_sec=20)
+            print(f"[oom-recover] lightweight probe done status={request_status}")
+    except Exception:
+        post_mix_status = None
+        request_status = None
+    elapsed = time.monotonic() - start
+    assert elapsed < 200, f"post-fault request should not hang after OOM removal: {elapsed:.2f}s"
+
+    if terminal_health == 200:
+        assert post_mix_status == 200, (
+            "health recovered but repeated mix multimodal request did not return HTTP 200; "
+            f"status={post_mix_status}, body_prefix={post_mix_body[:600]!r}"
+        )
+        recover_err = extract_openai_error_contract_from_bytes(post_mix_body)
+        assert recover_err is None, f"unexpected error object after recovery: {recover_err!r}"
+    else:
+        assert request_status is None or request_status >= 500, (
+            "unhealthy terminal state should fail fast on requests, "
+            f"got health={terminal_health}, request_status={request_status}"
+        )

tests/dfx/reliability/test_reliability_wan22.py

@@ -0,0 +1,411 @@
+"""
+Wan2.2 reliability integration tests.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+import pytest
+import requests
+
+from tests.dfx.conftest import (
+    assert_fault_exception,
+    create_reliability_omni_server_params,
+    resolve_oom_device_spec,
+    supports_video_generation,
+)
+from tests.dfx.reliability.helpers import (
+    get_health_raw,
+    inject_gpu_oom,
+    make_process_kill_fault_injector,
+    stop_gpu_oom_hogs,
+)
+from tests.helpers.mark import hardware_test
+from tests.helpers.media import generate_synthetic_image
+
+RELIABILITY_SCENARIOS: list[dict[str, Any]] = [
+    {
+        "test_name": "wan22_t2v_reliability_default",
+        "server_params": {
+            "model": "Wan-AI/Wan2.2-T2V-A14B-Diffusers",
+            "server_args": [
+                "--num-gpus",
+                "1",
+                "--boundary-ratio",
+                "0.875",
+                "--flow-shift",
+                "5.0",
+                "--disable-log-stats",
+            ],
+        },
+    }
+]
+
+E2E_STAGE_CONFIGS_DIR = Path(__file__).resolve().parent.parent / "e2e" / "stage_configs"
+OOM_INJECTION_CONFIG = {
+    "target_mem_ratio": 0.95,
+    "hold_seconds": 0,
+    "startup_timeout_sec": 20,
+    "strict": True,
+}
+FAULT_ERROR_KEYWORDS = (
+    "oom",
+    "out of memory",
+    "cuda",
+    "job failed",
+    "unknown error",
+    "internal server error",
+    "500 server error",
+)
+PROCESS_KILL_ERROR_KEYWORDS = (
+    "timeout",
+    "did not complete within",
+    "connection",
+    "engine",
+    "orchestrator",
+    "dead",
+    "internal",
+    "500",
+    "503",
+)
+
+WAN_PARAMS = create_reliability_omni_server_params(RELIABILITY_SCENARIOS, E2E_STAGE_CONFIGS_DIR)
+DIFFUSION_VIDEO_PARAMS = [param for param in WAN_PARAMS if supports_video_generation(param.model)]
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_gpu_oom_video_large_request_failure(omni_server_function, openai_client_function) -> None:
+    stage_config_path = getattr(omni_server_function, "stage_config_path", None)
+    device_spec = resolve_oom_device_spec(OOM_INJECTION_CONFIG, stage_config_path)
+    handle = inject_gpu_oom(
+        device=device_spec,
+        target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+        hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+        startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+        strict=OOM_INJECTION_CONFIG["strict"],
+    )
+    try:
+        image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+        request_config = {
+            "form_data": {
+                "prompt": "Generate a realistic road-driving video with camera motion.",
+                "width": 512,
+                "height": 512,
+                "fps": 8,
+                "num_frames": 8,
+                "guidance_scale": 1.0,
+                "flow_shift": 5.0,
+                "num_inference_steps": 4,
+                "seed": 42,
+            },
+            "image_reference": image_data_url,
+            "stream": False,
+        }
+        try:
+            openai_client_function.send_video_diffusion_request(request_config, request_num=1)
+        except Exception as exc:
+            assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+        else:
+            pytest.fail("expected large /v1/videos request failure during GPU OOM injection")
+    finally:
+        stop_gpu_oom_hogs(handle)
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+    "fault_injector",
+    [
+        pytest.param(
+            make_process_kill_fault_injector(
+                grep_patterns="multiprocessing.spawn",
+                signal_name="SIGKILL",
+                limit=1,
+                post_kill_wait_seconds=2.0,
+            ),
+            id="runtime_process_chain",
+        ),
+    ],
+    indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_video_request_failure(
+    omni_server_after_fault_function,
+    openai_client_function,
+) -> None:
+    image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+    request_config = {
+        "form_data": {
+            "prompt": "Generate a realistic road-driving video with camera motion.",
+            "width": 512,
+            "height": 512,
+            "fps": 8,
+            "num_frames": 8,
+            "guidance_scale": 1.0,
+            "flow_shift": 5.0,
+            "num_inference_steps": 4,
+            "seed": 42,
+        },
+        "image_reference": image_data_url,
+        "stream": False,
+    }
+    try:
+        openai_client_function.send_video_diffusion_request(request_config, request_num=1)
+    except Exception as exc:
+        assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS)
+    else:
+        pytest.fail("expected /v1/videos request failure after process-kill injection")
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+    "fault_injector",
+    [
+        pytest.param(
+            make_process_kill_fault_injector(
+                grep_patterns="multiprocessing.spawn",
+                signal_name="SIGKILL",
+                limit=1,
+                post_kill_wait_seconds=2.0,
+            ),
+            id="runtime_process_chain",
+        ),
+    ],
+    indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_video_health_fast_fail_and_concurrent(
+    omni_server_after_fault_function,
+    openai_client_function,
+) -> None:
+    """Black-box: after process kill, /v1/videos fails fast and concurrent calls don't hang.
+
+    /health→503 is checked last: optional ``pytest.skip`` then ``assert`` (drop the ``if`` to harden).
+    """
+    host = omni_server_after_fault_function.host
+    port = omni_server_after_fault_function.port
+    url = f"http://{host}:{port}/v1/videos"
+    image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+    request_config = {
+        "form_data": {
+            "prompt": "Generate a realistic road-driving video with camera motion.",
+            "width": 512,
+            "height": 512,
+            "fps": 8,
+            "num_frames": 8,
+            "guidance_scale": 1.0,
+            "flow_shift": 5.0,
+            "num_inference_steps": 4,
+            "seed": 42,
+        },
+        "image_reference": image_data_url,
+        "stream": False,
+    }
+
+    # Phase-1: one new /v1/videos request should fail fast.
+    payload = {
+        "prompt": "fast-fail check",
+        "width": "512",
+        "height": "512",
+        "fps": "8",
+        "num_frames": "8",
+        "num_inference_steps": "4",
+    }
+    start = time.monotonic()
+    try:
+        response = requests.post(
+            url,
+            data=payload,
+            headers={"Accept": "application/json"},
+            timeout=20,
+        )
+        elapsed = time.monotonic() - start
+        assert elapsed < 15, f"[process_kill fast_fail] /v1/videos did not fail fast after fault: {elapsed:.2f}s"
+        assert response.status_code >= 500, (
+            "[process_kill fast_fail] expected server-side error after fatal fault, "
+            f"got status={response.status_code}, body={response.text[:200]!r}"
+        )
+    except Exception:
+        elapsed = time.monotonic() - start
+        assert elapsed < 15, f"[process_kill fast_fail] /v1/videos exception was too slow after fault: {elapsed:.2f}s"
+
+    # Phase-2: concurrent requests should converge and at least one should fail.
+    start = time.monotonic()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
+        futures = [
+            executor.submit(openai_client_function.send_video_diffusion_request, request_config, 1) for _ in range(3)
+        ]
+        done, pending = concurrent.futures.wait(
+            futures,
+            timeout=40,
+            return_when=concurrent.futures.ALL_COMPLETED,
+        )
+
+    elapsed = time.monotonic() - start
+    assert not pending, f"[process_kill concurrent] some fault-time video requests hung: pending={len(pending)}"
+    assert elapsed < 40, f"[process_kill concurrent] fault-time video request convergence is too slow: {elapsed:.2f}s"
+
+    fault_observed = False
+    conc_debug: list[Any] = []
+    for future in done:
+        try:
+            future.result()
+            conc_debug.append("ok")
+        except Exception as exc:
+            fault_observed = True
+            conc_debug.append(repr(exc))
+            assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS)
+    assert fault_observed, (
+        "[process_kill concurrent] expected at least one /v1/videos request to fail after fault; "
+        f"conc_debug={conc_debug}"
+    )
+
+    # Phase-3: /health→503 after fault. Optional skip today; remove the ``if`` later to harden.
+    deadline = time.monotonic() + 20.0
+    last_observation = ""
+    final_health_status: int | None = None
+    while time.monotonic() < deadline:
+        try:
+            status, body = get_health_raw(host, port, timeout_sec=5)
+            last_observation = f"http={status}, body={body[:200]!r}"
+            final_health_status = status
+            if status == 503:
+                break
+        except Exception as exc:  # noqa: BLE001
+            last_observation = f"exception={exc!r}"
+        time.sleep(0.5)
+    if final_health_status != 503:
+        pytest.skip("issue#3050")
+    assert final_health_status == 503, (
+        "[process_kill health] expected /health 503 after fatal fault, "
+        f"got status={final_health_status}, last_observation={last_observation}"
+    )
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.skip(reason="issue#2327")
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_video_oom_recovers_after_fault_removed(
+    omni_server_function,
+) -> None:
+    """Black-box: after removing OOM pressure, service converges to a terminal state and requests do not hang."""
+    stage_config_path = getattr(omni_server_function, "stage_config_path", None)
+    device_spec = resolve_oom_device_spec(OOM_INJECTION_CONFIG, stage_config_path)
+    handle = inject_gpu_oom(
+        device=device_spec,
+        target_mem_ratio=OOM_INJECTION_CONFIG["target_mem_ratio"],
+        hold_seconds=OOM_INJECTION_CONFIG["hold_seconds"],
+        startup_timeout_sec=OOM_INJECTION_CONFIG["startup_timeout_sec"],
+        strict=OOM_INJECTION_CONFIG["strict"],
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    create_url = f"http://{host}:{port}/v1/videos"
+
+    failure_observed = False
+    try:
+        for _ in range(3):
+            try:
+                response = requests.post(
+                    create_url,
+                    data={
+                        "prompt": "oom recover probe",
+                        "width": "512",
+                        "height": "512",
+                        "fps": "8",
+                        "num_frames": "8",
+                        "num_inference_steps": "4",
+                    },
+                    headers={"Accept": "application/json"},
+                    timeout=25,
+                )
+                if response.status_code >= 500:
+                    failure_observed = True
+                    break
+            except Exception:
+                failure_observed = True
+                break
+            time.sleep(1.0)
+    finally:
+        stop_gpu_oom_hogs(handle)
+
+    assert failure_observed, "expected at least one video request failure while OOM pressure is active"
+
+    recovery_deadline = time.monotonic() + 90.0
+    terminal_health: int | None = None
+    unreachable_streak = 0
+    last_health_exc: BaseException | None = None
+    while time.monotonic() < recovery_deadline:
+        try:
+            status, _ = get_health_raw(host, port, timeout_sec=5)
+            unreachable_streak = 0
+            if status in (200, 503):
+                terminal_health = status
+                break
+        except Exception as exc:  # noqa: BLE001
+            last_health_exc = exc
+            msg = str(exc).lower()
+            if "connection refused" in msg or "actively refused" in msg:
+                unreachable_streak += 1
+                if unreachable_streak >= 5:
+                    pytest.fail(
+                        "after OOM sidecar stopped, /health is unreachable repeatedly; "
+                        "the APIServer process likely exited unexpectedly "
+                        f"(last_exc={last_health_exc!r})"
+                    )
+            else:
+                unreachable_streak = 0
+        time.sleep(1.0)
+    else:
+        pytest.fail(
+            "wan22 server did not converge to terminal health (200/503) after OOM pressure was removed; "
+            f"last_health_exc={last_health_exc!r}"
+        )
+
+    assert terminal_health is not None
+    start = time.monotonic()
+    recovery_resp: requests.Response | None = None
+    try:
+        recovery_resp = requests.post(
+            create_url,
+            data={
+                "prompt": "post-recovery admission check",
+                "width": "512",
+                "height": "512",
+                "fps": "8",
+                "num_frames": "8",
+                "num_inference_steps": "4",
+            },
+            headers={"Accept": "application/json"},
+            timeout=30,
+        )
+        recovery_status = recovery_resp.status_code
+        recovery_text_prefix = recovery_resp.text[:300]
+    except Exception as exc:  # noqa: BLE001
+        recovery_status = None
+        recovery_text_prefix = repr(exc)
+    elapsed = time.monotonic() - start
+    assert elapsed < 30, f"post-fault /v1/videos should not hang after OOM removal: {elapsed:.2f}s"
+
+    if terminal_health == 200:
+        assert recovery_status == 200, (
+            "health recovered but /v1/videos admission did not succeed; "
+            f"status={recovery_status}, body={recovery_text_prefix!r}"
+        )
+        assert recovery_resp is not None
+        recovery_payload = recovery_resp.json()
+        assert "id" in recovery_payload, f"post-recovery create payload missing id: {recovery_payload!r}"
+    else:
+        assert recovery_status is None or recovery_status >= 500, (
+            "unhealthy terminal state should fail fast on /v1/videos, "
+            f"got health={terminal_health}, request_status={recovery_status}, body={recovery_text_prefix!r}"
+        )

tests/helpers/fixtures/runtime.py

@@ -10,17 +10,27 @@ from __future__ import annotations
 
 import threading
 from collections.abc import Generator
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import pytest
-import yaml
 
-from tests.helpers.runtime import OmniServer
-from tests.helpers.stage_config import modify_stage_config
+if TYPE_CHECKING:
+    from tests.helpers.runtime import OmniServer
 
 omni_fixture_lock = threading.Lock()
 
 
+@pytest.fixture(scope="function")
+def omni_server_function(
+    request: pytest.FixtureRequest,
+    run_level: str,
+    model_prefix: str,
+) -> Generator[OmniServer, Any, None]:
+    from tests.helpers.runtime import run_omni_server
+
+    yield from run_omni_server(request, run_level, model_prefix, omni_fixture_lock)
+
+
 @pytest.fixture(scope="module")
 def omni_server(request: pytest.FixtureRequest, run_level: str, model_prefix: str) -> Generator[OmniServer, Any, None]:
     """Start vLLM-Omni through the standard or stage-CLI launcher.
@@ -29,81 +39,9 @@ def omni_server(request: pytest.FixtureRequest, run_level: str, model_prefix: st
     The ``use_stage_cli`` flag on ``OmniServerParams`` routes the setup through the
     stage-CLI harness while still reusing the same fixture grouping semantics.
     """
-    with omni_fixture_lock:
-        from tests.helpers.runtime import OmniServer, OmniServerParams, OmniServerStageCli
+    from tests.helpers.runtime import run_omni_server
 
-        params: OmniServerParams = request.param
-        model = model_prefix + params.model
-        port = params.port
-        stage_config_path = params.stage_config_path
-        if run_level in {"advanced_model", "full_model"} and stage_config_path is not None:
-            with open(stage_config_path, encoding="utf-8") as f:
-                cfg = yaml.safe_load(f) or {}
-            # Strip ``load_format: dummy`` (CI overlay default) so advanced_model
-            # tests use real weights. New schema (``stages:``) writes the field
-            # flat at stage level; legacy schema (``stage_args:``) nests it as
-            # ``engine_args.load_format``. Handle both.
-            new_schema_stages = cfg.get("stages")
-            stage_key = "stages" if new_schema_stages is not None else "stage_args"
-            delete_path = "load_format" if new_schema_stages is not None else "engine_args.load_format"
-            stage_entries = cfg.get(stage_key, [])
-            stage_ids = [stage["stage_id"] for stage in stage_entries if "stage_id" in stage]
-            stage_config_path = modify_stage_config(
-                stage_config_path,
-                deletes={stage_key: {stage_id: [delete_path] for stage_id in stage_ids}},
-            )
-
-        server_args = params.server_args or []
-        if params.use_omni and params.stage_init_timeout is not None:
-            server_args = [*server_args, "--stage-init-timeout", str(params.stage_init_timeout)]
-        else:
-            server_args = [*server_args, "--stage-init-timeout", "600"]
-        if params.init_timeout is not None:
-            server_args = [*server_args, "--init-timeout", str(params.init_timeout)]
-        else:
-            server_args = [*server_args, "--init-timeout", "900"]
-        if params.use_stage_cli:
-            if not params.use_omni:
-                raise ValueError("omni_server with use_stage_cli=True requires use_omni=True")
-            if stage_config_path is None:
-                raise ValueError("omni_server with use_stage_cli=True requires a stage_config_path")
-            server_args += ["--stage-configs-path", stage_config_path]
-
-            with OmniServerStageCli(
-                model,
-                stage_config_path,
-                server_args,
-                port=port,
-                env_dict=params.env_dict,
-            ) as server:
-                print("OmniServer started successfully")
-                yield server
-                print("OmniServer stopping...")
-        else:
-            if stage_config_path is not None:
-                server_args += ["--stage-configs-path", stage_config_path]
-
-            with (
-                OmniServer(
-                    model,
-                    server_args,
-                    port=port,
-                    env_dict=params.env_dict,
-                    use_omni=params.use_omni,
-                )
-                if port
-                else OmniServer(
-                    model,
-                    server_args,
-                    env_dict=params.env_dict,
-                    use_omni=params.use_omni,
-                )
-            ) as server:
-                print("OmniServer started successfully")
-                yield server
-                print("OmniServer stopping...")
-
-        print("OmniServer stopped")
+    yield from run_omni_server(request, run_level, model_prefix, omni_fixture_lock)
 
 
 @pytest.fixture
@@ -115,6 +53,15 @@ def openai_client(request: pytest.FixtureRequest, run_level: str):
     return OpenAIClientHandler(host=server.host, port=server.port, api_key="EMPTY", run_level=run_level)
 
 
+@pytest.fixture
+def openai_client_function(request: pytest.FixtureRequest, run_level: str):
+    """Resolve ``omni_server_function`` lazily for function-scoped reliability tests."""
+    from tests.helpers.runtime import OpenAIClientHandler
+
+    server = request.getfixturevalue("omni_server_function")
+    return OpenAIClientHandler(host=server.host, port=server.port, api_key="EMPTY", run_level=run_level)
+
+
 @pytest.fixture(scope="module")
 def omni_runner(request: pytest.FixtureRequest, model_prefix: str):
     from tests.helpers.runtime import OmniRunner

tests/helpers/runtime.py

@@ -9,7 +9,9 @@ import socket
 import subprocess
 import sys
 import tempfile
+import threading
 import time
+from collections.abc import Generator
 from dataclasses import dataclass
 from io import BytesIO
 from pathlib import Path
@@ -120,6 +122,89 @@ class OmniServerParams(NamedTuple):
     stage_init_timeout: int | None = None  # None: fixture supplies default (600 s)
 
 
+def run_omni_server(
+    request: Any,
+    run_level: str,
+    model_prefix: str,
+    omni_fixture_lock: threading.Lock,
+) -> Generator["OmniServer", Any, None]:
+    from tests.helpers.stage_config import modify_stage_config
+
+    with omni_fixture_lock:
+        params: OmniServerParams = request.param
+        model = model_prefix + params.model
+        port = params.port
+        stage_config_path = params.stage_config_path
+        if run_level in {"advanced_model", "full_model"} and stage_config_path is not None:
+            with open(stage_config_path, encoding="utf-8") as f:
+                cfg = yaml.safe_load(f) or {}
+            # Strip ``load_format: dummy`` (CI overlay default) so advanced_model
+            # tests use real weights. New schema (``stages:``) writes the field
+            # flat at stage level; legacy schema (``stage_args:``) nests it as
+            # ``engine_args.load_format``. Handle both.
+            new_schema_stages = cfg.get("stages")
+            stage_key = "stages" if new_schema_stages is not None else "stage_args"
+            delete_path = "load_format" if new_schema_stages is not None else "engine_args.load_format"
+            stage_entries = cfg.get(stage_key, [])
+            stage_ids = [stage["stage_id"] for stage in stage_entries if "stage_id" in stage]
+            stage_config_path = modify_stage_config(
+                stage_config_path,
+                deletes={stage_key: {stage_id: [delete_path] for stage_id in stage_ids}},
+            )
+
+        server_args = params.server_args or []
+        if params.use_omni and params.stage_init_timeout is not None:
+            server_args = [*server_args, "--stage-init-timeout", str(params.stage_init_timeout)]
+        else:
+            server_args = [*server_args, "--stage-init-timeout", "600"]
+        if params.init_timeout is not None:
+            server_args = [*server_args, "--init-timeout", str(params.init_timeout)]
+        else:
+            server_args = [*server_args, "--init-timeout", "900"]
+        if params.use_stage_cli:
+            if not params.use_omni:
+                raise ValueError("omni_server with use_stage_cli=True requires use_omni=True")
+            if stage_config_path is None:
+                raise ValueError("omni_server with use_stage_cli=True requires a stage_config_path")
+            server_args += ["--stage-configs-path", stage_config_path]
+
+            with OmniServerStageCli(
+                model,
+                stage_config_path,
+                server_args,
+                port=port,
+                env_dict=params.env_dict,
+            ) as server:
+                print("OmniServer started successfully")
+                yield server
+                print("OmniServer stopping...")
+        else:
+            if stage_config_path is not None:
+                server_args += ["--stage-configs-path", stage_config_path]
+
+            with (
+                OmniServer(
+                    model,
+                    server_args,
+                    port=port,
+                    env_dict=params.env_dict,
+                    use_omni=params.use_omni,
+                )
+                if port
+                else OmniServer(
+                    model,
+                    server_args,
+                    env_dict=params.env_dict,
+                    use_omni=params.use_omni,
+                )
+            ) as server:
+                print("OmniServer started successfully")
+                yield server
+                print("OmniServer stopping...")
+
+        print("OmniServer stopped")
+
+
 class OmniServer:
     """Omniserver for vLLM-Omni tests."""