[Test] Add scenarios for L5 reliability test (#3729)

Signed-off-by: zhumingjue <zhumingjue@huawei.com>

tests/dfx/reliability/helpers.py

@@ -4,10 +4,16 @@ 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
+
+Worker PID classification (``worker_pids`` in snapshots) is still used for logging /
+markers; post-fault **process** cleanup assertions use the full captured
+``tree_pids`` (serve root + descendants at fault time) — see
+:func:`assert_no_server_tree_process_residual_and_gpu_release`.
 """
 
 from __future__ import annotations
 
+import concurrent.futures
 import http.client
 import json
 import logging
@@ -27,6 +33,16 @@ import pytest
 
 logger = logging.getLogger(__name__)
 
+# Substrings matched against ``psutil.Process.name`` + argv text for classifying GPU
+# workers under the test server tree. Extend per-suite via
+# ``server.reliability_worker_markers_extra`` or replace via
+# ``server.reliability_worker_markers`` (see ``_resolve_runtime_worker_markers``).
+DEFAULT_RUNTIME_WORKER_MARKERS: tuple[str, ...] = (
+    "multiprocessing.spawn",
+    # Covers ``VLLM::Worker``, ``VLLM::StageEngineCoreProc_*`` (Omni stage engines), etc.
+    "VLLM::",
+)
+
 
 @dataclass
 class OomHandle:
@@ -452,6 +468,61 @@ def _safe_proc_info(pid: int) -> tuple[str, str]:
         return "<unknown>", "<unavailable>"
 
 
+def _normalize_worker_marker_substrings(seq: Sequence[Any], *, context: str) -> tuple[str, ...]:
+    """Return deduped non-empty marker substrings; raises if any entry is blank."""
+    out: list[str] = []
+    for item in seq:
+        s = str(item).strip()
+        if not s:
+            raise ValueError(f"{context}: worker marker must be a non-empty string, got {item!r}")
+        out.append(s)
+    return tuple(dict.fromkeys(out))
+
+
+def _resolve_runtime_worker_markers(
+    server: Any,
+    injector_markers_extra: Sequence[str] | None,
+) -> tuple[str, ...]:
+    """Resolve which substrings classify worker PIDs for fault snapshots.
+
+    If ``server.reliability_worker_markers`` is set to a sequence, **only** those
+    markers are used (no defaults). Otherwise defaults from
+    :data:`DEFAULT_RUNTIME_WORKER_MARKERS` are merged with:
+
+    - ``server.reliability_worker_markers_extra`` when present, and
+    - ``injector_markers_extra`` from :func:`make_server_root_kill_fault_injector` /
+      :func:`make_server_tree_kill_fault_injector`.
+    """
+    replace = getattr(server, "reliability_worker_markers", None)
+    if replace is not None:
+        return _normalize_worker_marker_substrings(replace, context="server.reliability_worker_markers")
+
+    merged: list[str] = []
+    merged.extend(DEFAULT_RUNTIME_WORKER_MARKERS)
+    extra_attr = getattr(server, "reliability_worker_markers_extra", None)
+    if extra_attr is not None:
+        merged.extend(
+            _normalize_worker_marker_substrings(extra_attr, context="server.reliability_worker_markers_extra")
+        )
+    if injector_markers_extra is not None:
+        merged.extend(
+            _normalize_worker_marker_substrings(
+                injector_markers_extra,
+                context="injector worker_markers_extra",
+            )
+        )
+    return tuple(dict.fromkeys(merged))
+
+
+def _pid_looks_like_runtime_worker(pid: int, markers: Sequence[str]) -> bool:
+    """True when process name or argv contains any resolved worker marker substring."""
+    if not markers:
+        return False
+    name, cmdline = _safe_proc_info(pid)
+    text = f"{name} {cmdline}"
+    return any(marker in text for marker in markers)
+
+
 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)
@@ -468,6 +539,29 @@ def _list_server_process_tree(server: Any) -> list[int]:
     return [root_pid, *descendants]
 
 
+def _pids_in_server_tree_substring_match(server: Any, pattern: str, *, limit: int) -> list[int]:
+    """Return up to ``limit`` PIDs in the server tree whose name+cmdline contains ``pattern`` literally.
+
+    ``pgrep -f`` matches the process command line used by procps and does not consult the
+    short process name; vLLM-Omni often exposes titles like ``VLLM::Worker`` or
+    ``VLLM::StageEngineCoreProc_*`` via prctl/setproctitle in ``ps``/``name()`` only.
+    This helper keeps kill injection aligned with :func:`_safe_proc_info` / fault snapshots.
+    """
+    if limit <= 0:
+        return []
+    tree = _list_server_process_tree(server)
+    if not tree or not pattern:
+        return []
+    hits: list[int] = []
+    for pid in tree:
+        if len(hits) >= limit:
+            break
+        name, cmdline = _safe_proc_info(int(pid))
+        if pattern in f"{name} {cmdline}":
+            hits.append(int(pid))
+    return hits
+
+
 def _log_server_process_tree(server: Any) -> None:
     """Print server process tree for debugging fault injection targets."""
     pids = _list_server_process_tree(server)
@@ -486,6 +580,315 @@ FaultInjector = Callable[[Any], None]
 """Callable invoked with the live ``OmniServer`` after it is ready (see ``omni_server_after_fault``)."""
 
 
+def run_fault_injection_with_rate_load(
+    *,
+    submit_request: Callable[[], Any],
+    inject_fault: Callable[[], None],
+    num_requests: int = 10,
+    request_rate: float = 0.3,
+    submit_interval_sec: float | None = None,
+    completion_timeout_sec: float = 120.0,
+) -> dict[str, Any]:
+    """Submit requests with a fixed rate, then inject fault once an in-flight request is observed."""
+    if num_requests <= 0:
+        raise ValueError("num_requests must be > 0")
+    if request_rate <= 0:
+        raise ValueError("request_rate must be > 0")
+
+    interval_sec = submit_interval_sec if submit_interval_sec is not None else (1.0 / request_rate)
+    if interval_sec < 0:
+        raise ValueError("submit_interval_sec must be >= 0")
+
+    max_workers = min(max(2, num_requests), 8)
+    injected = False
+    futures: list[concurrent.futures.Future[Any]] = []
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        for idx in range(num_requests):
+            futures.append(executor.submit(submit_request))
+            in_flight = any(not future.done() for future in futures)
+            if in_flight and not injected:
+                inject_fault()
+                injected = True
+            if idx != num_requests - 1 and interval_sec > 0:
+                time.sleep(interval_sec)
+
+        if not injected:
+            in_flight = any(not future.done() for future in futures)
+            if in_flight:
+                inject_fault()
+                injected = True
+
+        if not injected:
+            pytest.skip(
+                "no in-flight request observed before fault injection; "
+                "increase request cost or request rate for this environment"
+            )
+
+        done, pending = concurrent.futures.wait(
+            futures,
+            timeout=completion_timeout_sec,
+            return_when=concurrent.futures.ALL_COMPLETED,
+        )
+        assert not pending, (
+            "some in-flight-load requests did not converge after fault injection: "
+            f"pending={len(pending)} done={len(done)}"
+        )
+    failure_observed = False
+    completed = 0
+    success = 0
+    exceptions = 0
+    for future in futures:
+        completed += 1
+        try:
+            future.result()
+            success += 1
+        except Exception:  # noqa: BLE001
+            exceptions += 1
+            failure_observed = True
+    return {
+        "num_submitted": num_requests,
+        "completed": completed,
+        "success": success,
+        "exceptions": exceptions,
+        "failure_observed": failure_observed,
+        "inflight_observed": True,
+    }
+
+
+def list_alive_pids(pids: Sequence[int]) -> list[int]:
+    """Return PIDs from ``pids`` that still exist in the kernel and are **not** zombies.
+
+    After SIGTERM/SIGKILL the serve child may exit before the test harness calls
+    ``Popen.wait()``; until reaped, ``psutil.pid_exists`` stays true for the defunct
+    slot. Those PIDs must not count as "residual server processes" for leak checks.
+    """
+    out: list[int] = []
+    for pid in pids:
+        pid_i = int(pid)
+        if not psutil.pid_exists(pid_i):
+            continue
+        try:
+            if psutil.Process(pid_i).status() == psutil.STATUS_ZOMBIE:
+                continue
+        except psutil.Error:
+            continue
+        out.append(pid_i)
+    return out
+
+
+def query_gpu_compute_pid_used_memory_mb() -> dict[int, int] | None:
+    """Query NVIDIA compute-process memory map (pid -> used MB).
+
+    Returns ``None`` when ``nvidia-smi`` is unavailable/unreadable on current host.
+    """
+    out = subprocess.run(
+        ["nvidia-smi", "--query-compute-apps=pid,used_memory", "--format=csv,noheader,nounits"],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if out.returncode != 0:
+        logger.warning("[reliability][gpu] nvidia-smi query failed: %s", out.stderr.strip())
+        return None
+
+    pid_to_mem_mb: dict[int, int] = {}
+    for line in out.stdout.splitlines():
+        row = line.strip()
+        if not row:
+            continue
+        pieces = [part.strip() for part in row.split(",", maxsplit=1)]
+        if len(pieces) != 2:
+            continue
+        pid_raw, mem_raw = pieces
+        if not pid_raw.isdigit():
+            continue
+        mem_digits = "".join(ch for ch in mem_raw if ch.isdigit())
+        if not mem_digits:
+            continue
+        pid_to_mem_mb[int(pid_raw)] = int(mem_digits)
+    return pid_to_mem_mb
+
+
+def worker_residual_timeout_after_kill_signal(signal_name: str) -> float:
+    """Wall-clock budget for fault-snapshot PIDs / GPU to clear after ``signal_name``."""
+    if signal_name == "SIGKILL":
+        return 30.0
+    if signal_name == "SIGINT":
+        return 120.0
+    if signal_name == "SIGTERM":
+        return 90.0
+    return 30.0
+
+
+def assert_no_server_tree_process_residual_and_gpu_release(
+    server: Any,
+    *,
+    scenario: str,
+    timeout_sec: float = 30.0,
+    poll_interval_sec: float = 0.5,
+) -> None:
+    """Assert no live PIDs remain from the fault-time server tree and no GPU use by those PIDs.
+
+    Uses ``reliability_fault_snapshot["tree_pids"]`` (root + all descendants captured at
+    injection time), not the marker-filtered ``worker_pids`` subset — so helpers like
+    ``multiprocessing.resource_tracker`` are included in the residual check.
+
+    **Zombie** PIDs (exited children not yet reaped by the test ``Popen``) are excluded
+    from the "alive" list; ``psutil.pid_exists`` alone would false-positive on them.
+
+    Note: if a child is reparented to PID 1 while staying alive, its PID is unchanged
+    and remains detected; if it **exits and a new unrelated process reuses the same
+    numeric PID**, this check may false-positive (rare on short windows).
+    """
+    snapshot = getattr(server, "reliability_fault_snapshot", None)
+    if not isinstance(snapshot, dict):
+        pytest.fail(f"[{scenario}] missing reliability fault snapshot on server")
+    tree_pids = [int(pid) for pid in snapshot.get("tree_pids", [])]
+    if not tree_pids:
+        pytest.skip(f"[{scenario}] no server process tree PIDs captured for this run")
+
+    tree_pid_set = set(tree_pids)
+    deadline = time.monotonic() + timeout_sec
+    last_alive: list[int] = []
+    last_gpu_leaks: dict[int, int] = {}
+    while time.monotonic() < deadline:
+        last_alive = list_alive_pids(tree_pids)
+        gpu_map = query_gpu_compute_pid_used_memory_mb()
+        if gpu_map is None:
+            pytest.skip(f"[{scenario}] nvidia-smi unavailable; skip GPU release assertion")
+        last_gpu_leaks = {pid: mem_mb for pid, mem_mb in gpu_map.items() if pid in tree_pid_set and mem_mb > 0}
+        if not last_alive and not last_gpu_leaks:
+            return
+        time.sleep(poll_interval_sec)
+
+    assert not last_alive, f"[{scenario}] residual server-tree processes remain alive: {last_alive}"
+    assert not last_gpu_leaks, f"[{scenario}] server-tree PID GPU memory not released: {last_gpu_leaks}"
+
+
+def assert_no_worker_residual_and_gpu_release(
+    server: Any,
+    *,
+    scenario: str,
+    timeout_sec: float = 30.0,
+    poll_interval_sec: float = 0.5,
+) -> None:
+    """Deprecated: use :func:`assert_no_server_tree_process_residual_and_gpu_release`."""
+    assert_no_server_tree_process_residual_and_gpu_release(
+        server,
+        scenario=scenario,
+        timeout_sec=timeout_sec,
+        poll_interval_sec=poll_interval_sec,
+    )
+
+
+def _capture_server_fault_snapshot(
+    server: Any,
+    *,
+    worker_markers_extra: Sequence[str] | None = None,
+) -> dict[str, list[int]]:
+    """Capture current server process snapshot for post-fault assertions.
+
+    ``tree_pids`` is ``[root, ...descendants]`` and is used by
+    :func:`assert_no_server_tree_process_residual_and_gpu_release`.
+    ``worker_pids`` is the marker-filtered subset for debugging only.
+    """
+    markers = _resolve_runtime_worker_markers(server, worker_markers_extra)
+    tree_pids = _list_server_process_tree(server)
+    root_pid = tree_pids[0] if tree_pids else None
+    worker_pids: list[int] = []
+    for pid in tree_pids:
+        if root_pid is not None and int(pid) == int(root_pid):
+            continue
+        if _pid_looks_like_runtime_worker(pid, markers):
+            worker_pids.append(pid)
+    snapshot = {
+        "root_pids": [root_pid] if root_pid is not None else [],
+        "tree_pids": tree_pids,
+        "worker_pids": worker_pids,
+        "worker_markers": list(markers),
+    }
+    setattr(server, "reliability_fault_snapshot", snapshot)
+    return snapshot
+
+
+def make_server_root_kill_fault_injector(
+    *,
+    signal_name: str = "SIGKILL",
+    post_kill_wait_seconds: float = 0.0,
+    worker_markers_extra: Sequence[str] | None = None,
+) -> FaultInjector:
+    """Kill only the current server root process (simulate first strike on PID1-like process).
+
+    ``worker_markers_extra`` is merged into worker PID detection for the fault snapshot
+    (after :data:`DEFAULT_RUNTIME_WORKER_MARKERS` unless
+    ``server.reliability_worker_markers`` replaces the set entirely).
+    """
+
+    def _inject(server: Any) -> None:
+        _log_server_process_tree(server)
+        snapshot = _capture_server_fault_snapshot(server, worker_markers_extra=worker_markers_extra)
+        root_pids = snapshot["root_pids"]
+        if not root_pids:
+            pytest.skip("no server root process found for root-kill injection")
+        root_pid = int(root_pids[0])
+        sig = getattr(signal, signal_name, None)
+        if sig is None:
+            raise ValueError(f"Unsupported signal_name: {signal_name}")
+        name, cmdline = _safe_proc_info(root_pid)
+        print(
+            f"[reliability][process-kill] root-kill pid={root_pid} name={name} signal={signal_name} cmdline={cmdline}",
+            flush=True,
+        )
+        os.kill(root_pid, sig)
+        if post_kill_wait_seconds > 0:
+            time.sleep(post_kill_wait_seconds)
+
+    return _inject
+
+
+def make_server_tree_kill_fault_injector(
+    *,
+    signal_name: str = "SIGKILL",
+    post_kill_wait_seconds: float = 0.0,
+    inter_kill_wait_seconds: float = 0.2,
+    worker_markers_extra: Sequence[str] | None = None,
+) -> FaultInjector:
+    """Kill current server root first, then kill all remaining descendants.
+
+    See :func:`make_server_root_kill_fault_injector` for ``worker_markers_extra``.
+    """
+
+    def _inject(server: Any) -> None:
+        _log_server_process_tree(server)
+        snapshot = _capture_server_fault_snapshot(server, worker_markers_extra=worker_markers_extra)
+        tree_pids = [int(pid) for pid in snapshot["tree_pids"]]
+        if not tree_pids:
+            pytest.skip("no server process tree found for tree-kill injection")
+
+        sig = getattr(signal, signal_name, None)
+        if sig is None:
+            raise ValueError(f"Unsupported signal_name: {signal_name}")
+
+        root_pid = tree_pids[0]
+        kill_order = [root_pid, *[pid for pid in tree_pids if pid != root_pid]]
+        for pid in kill_order:
+            if not psutil.pid_exists(pid):
+                continue
+            name, cmdline = _safe_proc_info(pid)
+            print(
+                f"[reliability][process-kill] tree-kill pid={pid} name={name} signal={signal_name} cmdline={cmdline}",
+                flush=True,
+            )
+            os.kill(pid, sig)
+            if inter_kill_wait_seconds > 0:
+                time.sleep(inter_kill_wait_seconds)
+
+        if post_kill_wait_seconds > 0:
+            time.sleep(post_kill_wait_seconds)
+
+    return _inject
+
+
 def make_process_kill_fault_injector(
     *,
     grep_patterns: str | Sequence[str],
@@ -493,11 +896,18 @@ def make_process_kill_fault_injector(
     limit: int = 1,
     post_kill_wait_seconds: float = 0.0,
 ) -> FaultInjector:
-    """Build a post-ready injector that kills processes matched by ``pgrep -f``.
+    """Build a post-ready injector that kills processes matched by ordered ``grep_patterns``.
 
-    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).
+    Matching is tried in two phases per pattern (first hit wins):
+
+    1. **Server process tree** (``psutil``): literal substring match against
+       ``Process.name()`` plus argv text. This matches how :func:`_safe_proc_info`
+       logs processes and catches vLLM-style titles (``VLLM::...``) that often do not
+       appear in ``pgrep -f``\'s command-line view.
+    2. **``pgrep -f``** (legacy): scoped to the server PID tree when it is known;
+       uses procps regular-expression rules for the pattern.
+
+    If neither phase finds a target, the returned callable issues ``pytest.skip``.
 
     Args:
         grep_patterns: One pattern or an ordered list of patterns.
@@ -514,9 +924,45 @@ def make_process_kill_fault_injector(
             logger.warning(
                 "[reliability][process-kill] no server process tree found; fallback to global pgrep matching"
             )
+
+        def _kill_and_wait(filtered: list[int], pattern: str, *, source: str) -> None:
+            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} "
+                    f"source={source} cmdline={cmdline}",
+                    flush=True,
+                )
+                os.kill(pid, sig)
+            print(
+                f"[reliability][process-kill] matched pattern={pattern!r} source={source} "
+                f"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)
+
         for pattern in patterns:
             print(
-                f"[reliability][process-kill] trying pattern={pattern} signal={signal_name} limit={limit}",
+                f"[reliability][process-kill] trying server_tree substring pattern={pattern!r} "
+                f"signal={signal_name} limit={limit}",
+                flush=True,
+            )
+            tree_hits = _pids_in_server_tree_substring_match(server, pattern, limit=limit)
+            if tree_hits:
+                _kill_and_wait(tree_hits, pattern, source="server_tree")
+                return
+
+        for pattern in patterns:
+            print(
+                f"[reliability][process-kill] trying pgrep -f pattern={pattern!r} signal={signal_name} limit={limit}",
                 flush=True,
             )
             pids = inject_process_kill(
@@ -535,26 +981,7 @@ def make_process_kill_fault_injector(
                 )
                 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)
+                _kill_and_wait(filtered, pattern, source="pgrep")
                 return
         logger.warning(
             "[reliability][process-kill] no process matched patterns=%s signal=%s limit=%s",

tests/dfx/reliability/test_reliability_qwen3_omni.py

@@ -21,13 +21,19 @@ from tests.dfx.conftest import (
     resolve_oom_device_spec,
 )
 from tests.dfx.reliability.helpers import (
+    FaultInjector,
+    assert_no_server_tree_process_residual_and_gpu_release,
     extract_openai_error_contract_from_bytes,
     get_health_raw,
     inject_gpu_oom,
     make_process_kill_fault_injector,
+    make_server_root_kill_fault_injector,
+    make_server_tree_kill_fault_injector,
     post_chat_completions_raw,
     post_json_raw,
+    run_fault_injection_with_rate_load,
     stop_gpu_oom_hogs,
+    worker_residual_timeout_after_kill_signal,
 )
 from tests.helpers.mark import hardware_test
 from tests.helpers.media import generate_synthetic_audio, generate_synthetic_image, generate_synthetic_video
@@ -97,7 +103,73 @@ FAULT_ERROR_KEYWORDS = (
     "500",
     "503",
 )
-RUNTIME_WORKER_PATTERN = "VLLM::Worker"
+PROCESS_KILL_ERROR_KEYWORDS = (
+    "timeout",
+    "did not complete within",
+    "connection",
+    "engine",
+    "orchestrator",
+    "dead",
+    "internal",
+    "500",
+    "503",
+)
+# Prefix match for vLLM-titled children (``VLLM::Worker``, ``VLLM::StageEngineCoreProc_*``, …).
+# Must be a literal substring of ``name``/``cmdline`` — no trailing whitespace (would break match).
+RUNTIME_WORKER_PATTERN = "VLLM::"
+# RFC#2366 signal x target matrix:
+# - worker: SIGTERM, SIGKILL
+# - serve-root no-load: SIGINT; SIGTERM/SIGKILL skipped (vllm issue#43060)
+# - serve-root with-load: SIGINT; SIGTERM skipped (issue#3683); SIGKILL skipped (issue#43060)
+# - serve-tree no-load: SIGTERM/SIGKILL
+# - serve-tree with-load: SIGKILL; SIGTERM skipped (issue#3683)
+_SERVE_ROOT_NON_SIGINT_SKIP = pytest.mark.skip(reason="vllm issue#43060")
+_WITH_LOAD_SIGTERM_SKIP = pytest.mark.skip(reason="issue#3683")
+SERVE_SIGNAL_PARAMS = [
+    pytest.param("SIGTERM", id="sigterm"),
+    pytest.param("SIGINT", id="sigint"),
+    pytest.param("SIGKILL", id="sigkill", marks=_SERVE_ROOT_NON_SIGINT_SKIP),
+]
+SERVE_WITH_LOAD_SIGNAL_PARAMS = [
+    pytest.param("SIGTERM", id="sigterm", marks=_WITH_LOAD_SIGTERM_SKIP),
+    pytest.param("SIGINT", id="sigint", marks=_WITH_LOAD_SIGTERM_SKIP),
+    pytest.param("SIGKILL", id="sigkill", marks=_WITH_LOAD_SIGTERM_SKIP),
+]
+TREE_SIGNAL_PARAMS = [
+    pytest.param("SIGTERM", id="sigterm"),
+    pytest.param("SIGKILL", id="sigkill"),
+]
+TREE_WITH_LOAD_SIGNAL_PARAMS = [
+    pytest.param("SIGTERM", id="sigterm", marks=_WITH_LOAD_SIGTERM_SKIP),
+    pytest.param("SIGKILL", id="sigkill"),
+]
+WORKER_SIGNAL_FAULT_PARAMS = [
+    pytest.param(
+        make_process_kill_fault_injector(
+            grep_patterns=RUNTIME_WORKER_PATTERN,
+            signal_name="SIGTERM",
+            limit=1,
+            post_kill_wait_seconds=2.0,
+        ),
+        id="runtime_process_chain_sigterm",
+    ),
+    pytest.param(
+        make_process_kill_fault_injector(
+            grep_patterns=RUNTIME_WORKER_PATTERN,
+            signal_name="SIGKILL",
+            limit=1,
+            post_kill_wait_seconds=2.0,
+        ),
+        id="runtime_process_chain_sigkill",
+    ),
+]
+INFLIGHT_INJECTION_REQUEST_RATE = 0.3
+INFLIGHT_INJECTION_REQUEST_COUNT = 10
+
+# Post-fault chat probe: ``MAX_ELAPSED`` aligns with HTTP read timeout; assert allows +1s jitter
+# when the client raises at timeout (e.g. ~30.03s).
+_PROCESS_KILL_WORKER_CHAT_FF_HTTP_TIMEOUT_SEC = 30
+_PROCESS_KILL_WORKER_CHAT_FF_MAX_ELAPSED_SEC = 30
 
 
 class _HasServeArgs(Protocol):
@@ -194,6 +266,55 @@ def _looks_like_server_unreachable(exc: BaseException) -> bool:
     return "connection refused" in msg or "actively refused" in msg
 
 
+def _chat_request_config(omni_server: _HasServeArgs) -> dict[str, Any]:
+    messages = dummy_messages_from_mix_data(
+        system_prompt=_get_system_prompt(),
+        content_text="What is the capital of China? Answer in one short sentence.",
+    )
+    return {
+        "model": omni_server.model,
+        "messages": messages,
+        "stream": False,
+        "modalities": ["text"],
+    }
+
+
+def _assert_post_fault_chat_fast_fail(host: str, port: int, *, model: str, scenario: str) -> None:
+    payload = {
+        "model": model,
+        "messages": [{"role": "user", "content": "post-fault fast-fail check"}],
+        "stream": False,
+        "modalities": ["text"],
+    }
+    start = time.monotonic()
+    try:
+        status, body = post_json_raw(host, port, "/v1/chat/completions", payload, timeout_sec=20)
+        elapsed = time.monotonic() - start
+        assert elapsed < 15, f"[{scenario} fast_fail] /v1/chat/completions did not fail fast: {elapsed:.2f}s"
+        assert status >= 500, (
+            f"[{scenario} fast_fail] expected server-side error after fault, got status={status}, body={body[:200]!r}"
+        )
+    except Exception:  # noqa: BLE001
+        elapsed = time.monotonic() - start
+        assert elapsed < 15, f"[{scenario} fast_fail] exception was too slow after fault: {elapsed:.2f}s"
+
+
+def _assert_post_fault_health_terminal(host: str, port: int, *, scenario: str) -> None:
+    deadline = time.monotonic() + 20.0
+    last_observation = ""
+    while time.monotonic() < deadline:
+        try:
+            status, body = get_health_raw(host, port, timeout_sec=5)
+            last_observation = f"http={status}, body={body[:200]!r}"
+            if status == 503:
+                return
+        except Exception as exc:  # noqa: BLE001
+            last_observation = f"exception={exc!r}"
+            return
+        time.sleep(0.5)
+    pytest.fail(f"[{scenario} health] no terminal post-fault health observed: {last_observation}")
+
+
 QWEN_PARAMS = create_reliability_omni_server_params(RELIABILITY_SCENARIOS, DEPLOY_CONFIGS_DIR)
 
 
@@ -354,17 +475,7 @@ def test_reliability_fault_gpu_oom_concurrent_pressure_failure(omni_server_funct
 @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",
-        ),
-    ],
+    WORKER_SIGNAL_FAULT_PARAMS,
     indirect=True,
 )
 @pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
@@ -385,7 +496,7 @@ def test_reliability_fault_process_kill_request_failure(
     try:
         openai_client_function.send_omni_request(request_config, request_num=1)
     except Exception as exc:
-        assert_fault_exception(exc, FAULT_ERROR_KEYWORDS)
+        assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS)
     else:
         pytest.fail("expected request failure after process-kill injection")
 
@@ -394,48 +505,22 @@ def test_reliability_fault_process_kill_request_failure(
 @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",
-        ),
-    ],
+    WORKER_SIGNAL_FAULT_PARAMS,
     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."""
+    """Black-box: after worker process kill, chat fails fast, concurrent chat does not hang; /health→503 last.
+
+    ``SIGTERM`` on a stage engine often exceeds a 15s client-visible path; ``SIGKILL`` is
+    typically quicker. See module constants ``_PROCESS_KILL_WORKER_CHAT_FF_*``.
+    """
     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."}],
@@ -444,20 +529,28 @@ def test_reliability_fault_process_kill_health_fast_fail_and_concurrent(
     }
     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)
+        ff_status, ff_body = post_json_raw(
+            host,
+            port,
+            "/v1/chat/completions",
+            payload,
+            timeout_sec=_PROCESS_KILL_WORKER_CHAT_FF_HTTP_TIMEOUT_SEC,
+        )
         elapsed = time.monotonic() - start
-        assert elapsed < 15, f"[process_kill fast_fail] request did not fail fast after fault: {elapsed:.2f}s"
+        assert elapsed <= _PROCESS_KILL_WORKER_CHAT_FF_MAX_ELAPSED_SEC + 1.0, (
+            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
+    except Exception:
         elapsed = time.monotonic() - start
-        assert elapsed < 15, f"[process_kill fast_fail] request exception was too slow after fault: {elapsed:.2f}s"
+        assert elapsed <= _PROCESS_KILL_WORKER_CHAT_FF_MAX_ELAPSED_SEC + 1.0, (
+            f"[process_kill fast_fail] request exception was too slow after fault: {elapsed:.2f}s"
+        )
 
     payload_json = json.dumps(
         {
@@ -490,32 +583,195 @@ def test_reliability_fault_process_kill_health_fast_fail_and_concurrent(
     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))
+            assert_fault_exception(exc, PROCESS_KILL_ERROR_KEYWORDS)
             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"
     )
 
+    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
-@pytest.mark.skip(reason="issue#2327")
+@pytest.mark.skip(reason="issue#3683")
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+    "fault_injector",
+    WORKER_SIGNAL_FAULT_PARAMS,
+    indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_worker_with_load_request_failure(
+    omni_server_function,
+    openai_client_function,
+    fault_injector: FaultInjector,
+) -> None:
+    request_config = _chat_request_config(omni_server_function)
+    scenario = "kill_worker_with_load"
+    load_result = run_fault_injection_with_rate_load(
+        submit_request=lambda: openai_client_function.send_omni_request(request_config, request_num=1),
+        inject_fault=lambda: fault_injector(omni_server_function),
+        num_requests=INFLIGHT_INJECTION_REQUEST_COUNT,
+        request_rate=INFLIGHT_INJECTION_REQUEST_RATE,
+        completion_timeout_sec=120.0,
+    )
+    assert load_result["failure_observed"], (
+        f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}"
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", SERVE_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_serve_root_no_load_fast_fail_and_cleanup(
+    omni_server_function,
+    signal_name: str,
+) -> None:
+    """Black-box: kill serve root without load; verify fast-fail/health/cleanup."""
+    scenario = f"kill_serve_root_no_load_{signal_name.lower()}"
+    injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0)
+    injector(omni_server_function)
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=2)
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", SERVE_WITH_LOAD_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_serve_root_with_load_fast_fail_and_cleanup(
+    omni_server_function,
+    openai_client_function,
+    signal_name: str,
+) -> None:
+    request_config = _chat_request_config(omni_server_function)
+    scenario = f"kill_serve_root_with_load_{signal_name.lower()}"
+    injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0)
+    load_result = run_fault_injection_with_rate_load(
+        submit_request=lambda: openai_client_function.send_omni_request(request_config, request_num=1),
+        inject_fault=lambda: injector(omni_server_function),
+        num_requests=INFLIGHT_INJECTION_REQUEST_COUNT,
+        request_rate=INFLIGHT_INJECTION_REQUEST_RATE,
+        completion_timeout_sec=300.0,
+    )
+    assert load_result["failure_observed"], (
+        f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}"
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", TREE_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_tree_no_load_fast_fail_and_cleanup(
+    omni_server_function,
+    signal_name: str,
+) -> None:
+    """Black-box: kill server tree without load; verify fast-fail/health/cleanup."""
+    scenario = f"kill_serve_tree_no_load_{signal_name.lower()}"
+    injector = make_server_tree_kill_fault_injector(
+        signal_name=signal_name,
+        post_kill_wait_seconds=2.0,
+        inter_kill_wait_seconds=0.1,
+    )
+    injector(omni_server_function)
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", TREE_WITH_LOAD_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", QWEN_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_tree_with_load_fast_fail_and_cleanup(
+    omni_server_function,
+    openai_client_function,
+    signal_name: str,
+) -> None:
+    request_config = _chat_request_config(omni_server_function)
+    scenario = f"kill_serve_tree_with_load_{signal_name.lower()}"
+    injector = make_server_tree_kill_fault_injector(
+        signal_name=signal_name,
+        post_kill_wait_seconds=2.0,
+        inter_kill_wait_seconds=0.1,
+    )
+    load_result = run_fault_injection_with_rate_load(
+        submit_request=lambda: openai_client_function.send_omni_request(request_config, request_num=1),
+        inject_fault=lambda: injector(omni_server_function),
+        num_requests=INFLIGHT_INJECTION_REQUEST_COUNT,
+        request_rate=INFLIGHT_INJECTION_REQUEST_RATE,
+        completion_timeout_sec=120.0,
+    )
+    assert load_result["failure_observed"], (
+        f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}"
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_chat_fast_fail(host, port, model=omni_server_function.model, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.skip(reason="OOM injection causes the serving instance to exit")
 @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(

tests/dfx/reliability/test_reliability_wan22.py

@@ -20,28 +20,25 @@ from tests.dfx.conftest import (
     supports_video_generation,
 )
 from tests.dfx.reliability.helpers import (
+    FaultInjector,
+    assert_no_server_tree_process_residual_and_gpu_release,
     get_health_raw,
     inject_gpu_oom,
     make_process_kill_fault_injector,
+    make_server_root_kill_fault_injector,
+    make_server_tree_kill_fault_injector,
+    run_fault_injection_with_rate_load,
     stop_gpu_oom_hogs,
+    worker_residual_timeout_after_kill_signal,
 )
 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",
+        "test_name": "wan22_i2v_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",
-            ],
+            "model": "Wan-AI/Wan2.2-I2V-A14B-Diffusers",
         },
     }
 ]
@@ -73,9 +70,114 @@ PROCESS_KILL_ERROR_KEYWORDS = (
     "500",
     "503",
 )
+# RFC#2366 signal x target matrix (aligned with test_reliability_qwen3_omni.py):
+# - worker: SIGTERM, SIGKILL
+# - serve-root: SIGTERM, SIGINT; SIGKILL skipped (issue#3725)
+# - serve-tree no-load: SIGTERM, SIGKILL
+# - serve-tree with-load: SIGTERM, SIGKILL
+_SERVE_ROOT_SIGKILL_SKIP = pytest.mark.skip(reason="issue#3725")
+SERVE_SIGNAL_PARAMS = [
+    pytest.param("SIGTERM", id="sigterm"),
+    pytest.param("SIGINT", id="sigint"),
+    pytest.param("SIGKILL", id="sigkill", marks=_SERVE_ROOT_SIGKILL_SKIP),
+]
+TREE_SIGNAL_PARAMS = [
+    pytest.param("SIGTERM", id="sigterm"),
+    pytest.param("SIGKILL", id="sigkill"),
+]
+TREE_WITH_LOAD_SIGNAL_PARAMS = [
+    pytest.param("SIGTERM", id="sigterm"),
+    pytest.param("SIGKILL", id="sigkill"),
+]
+WORKER_SIGNAL_FAULT_PARAMS = [
+    pytest.param(
+        make_process_kill_fault_injector(
+            grep_patterns="vLLM-Omni::",
+            signal_name="SIGTERM",
+            limit=1,
+            post_kill_wait_seconds=2.0,
+        ),
+        id="runtime_process_chain_sigterm",
+    ),
+    pytest.param(
+        make_process_kill_fault_injector(
+            grep_patterns="vLLM-Omni::",
+            signal_name="SIGKILL",
+            limit=1,
+            post_kill_wait_seconds=2.0,
+        ),
+        id="runtime_process_chain_sigkill",
+    ),
+]
 
 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)]
+INFLIGHT_INJECTION_REQUEST_RATE = 0.3
+INFLIGHT_INJECTION_REQUEST_COUNT = 10
+
+
+def _video_request_config() -> dict[str, Any]:
+    image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(1280, 720)['base64']}"
+    return {
+        "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,
+    }
+
+
+def _assert_post_fault_video_fast_fail(host: str, port: int, *, scenario: str) -> None:
+    url = f"http://{host}:{port}/v1/videos"
+    payload = {
+        "prompt": "post-fault 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"[{scenario} fast_fail] /v1/videos did not fail fast: {elapsed:.2f}s"
+        assert response.status_code >= 500, (
+            f"[{scenario} fast_fail] expected server-side error after fault, "
+            f"got status={response.status_code}, body={response.text[:200]!r}"
+        )
+    except Exception:  # noqa: BLE001
+        elapsed = time.monotonic() - start
+        assert elapsed < 15, f"[{scenario} fast_fail] exception was too slow after fault: {elapsed:.2f}s"
+
+
+def _assert_post_fault_health_terminal(host: str, port: int, *, scenario: str) -> None:
+    deadline = time.monotonic() + 20.0
+    last_observation = ""
+    while time.monotonic() < deadline:
+        try:
+            status, body = get_health_raw(host, port, timeout_sec=5)
+            last_observation = f"http={status}, body={body[:200]!r}"
+            if status == 503:
+                return
+        except Exception as exc:  # noqa: BLE001
+            last_observation = f"exception={exc!r}"
+            return
+        time.sleep(0.5)
+    pytest.fail(f"[{scenario} health] no terminal post-fault health observed: {last_observation}")
 
 
 @pytest.mark.slow
@@ -122,40 +224,15 @@ def test_reliability_fault_gpu_oom_video_large_request_failure(omni_server_funct
 @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",
-        ),
-    ],
+    WORKER_SIGNAL_FAULT_PARAMS,
     indirect=True,
 )
 @pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
-def test_reliability_fault_process_kill_video_request_failure(
+def test_reliability_fault_process_kill_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,
-    }
+    request_config = _video_request_config()
     try:
         openai_client_function.send_video_diffusion_request(request_config, request_num=1)
     except Exception as exc:
@@ -168,21 +245,11 @@ def test_reliability_fault_process_kill_video_request_failure(
 @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",
-        ),
-    ],
+    WORKER_SIGNAL_FAULT_PARAMS,
     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(
+def test_reliability_fault_process_kill_health_fast_fail_and_concurrent(
     omni_server_after_fault_function,
     openai_client_function,
 ) -> None:
@@ -290,6 +357,37 @@ def test_reliability_fault_process_kill_video_health_fast_fail_and_concurrent(
     )
 
 
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize(
+    "fault_injector",
+    WORKER_SIGNAL_FAULT_PARAMS,
+    indirect=True,
+)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_worker_with_load_request_failure(
+    omni_server_function,
+    openai_client_function,
+    fault_injector: FaultInjector,
+) -> None:
+    request_config = _video_request_config()
+    scenario = "kill_worker_with_load"
+    load_result = run_fault_injection_with_rate_load(
+        submit_request=lambda: openai_client_function.send_video_diffusion_request(request_config, request_num=1),
+        inject_fault=lambda: fault_injector(omni_server_function),
+        num_requests=INFLIGHT_INJECTION_REQUEST_COUNT,
+        request_rate=INFLIGHT_INJECTION_REQUEST_RATE,
+        completion_timeout_sec=120.0,
+    )
+    assert load_result["failure_observed"], (
+        f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}"
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_video_fast_fail(host, port, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+
+
 @pytest.mark.slow
 @hardware_test(res={"cuda": "H100"}, num_cards=1)
 @pytest.mark.skip(reason="issue#2327")
@@ -409,3 +507,127 @@ def test_reliability_video_oom_recovers_after_fault_removed(
             "unhealthy terminal state should fail fast on /v1/videos, "
             f"got health={terminal_health}, request_status={recovery_status}, body={recovery_text_prefix!r}"
         )
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", SERVE_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_serve_root_with_load_fast_fail_and_cleanup(
+    omni_server_function,
+    openai_client_function,
+    signal_name: str,
+) -> None:
+    """Black-box: during in-flight video requests, kill serve root and verify post-fault behavior/cleanup."""
+    request_config = _video_request_config()
+    scenario = f"kill_serve_root_with_load_{signal_name.lower()}"
+    injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0)
+    load_result = run_fault_injection_with_rate_load(
+        submit_request=lambda: openai_client_function.send_video_diffusion_request(request_config, request_num=1),
+        inject_fault=lambda: injector(omni_server_function),
+        num_requests=INFLIGHT_INJECTION_REQUEST_COUNT,
+        request_rate=INFLIGHT_INJECTION_REQUEST_RATE,
+        completion_timeout_sec=120.0,
+    )
+    assert load_result["failure_observed"], (
+        f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}"
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_video_fast_fail(host, port, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )
+
+
+@pytest.mark.slow
+@hardware_test(res={"cuda": "H100"}, num_cards=1)
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", SERVE_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_serve_root_no_load_fast_fail_and_cleanup(
+    omni_server_function,
+    signal_name: str,
+) -> None:
+    """Black-box: kill serve root without load; verify fast-fail/health/cleanup."""
+    scenario = f"kill_serve_root_no_load_{signal_name.lower()}"
+    injector = make_server_root_kill_fault_injector(signal_name=signal_name, post_kill_wait_seconds=2.0)
+    injector(omni_server_function)
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_video_fast_fail(host, port, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", TREE_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_tree_no_load_fast_fail_and_cleanup(
+    omni_server_function,
+    signal_name: str,
+) -> None:
+    """Black-box: kill server tree without load; verify fast-fail/health/cleanup."""
+    scenario = f"kill_serve_tree_no_load_{signal_name.lower()}"
+    injector = make_server_tree_kill_fault_injector(
+        signal_name=signal_name,
+        post_kill_wait_seconds=2.0,
+        inter_kill_wait_seconds=0.1,
+    )
+    injector(omni_server_function)
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_video_fast_fail(host, port, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )
+
+
+@pytest.mark.slow
+@pytest.mark.skipif(os.name == "nt", reason="process-kill injection helper is POSIX-only")
+@pytest.mark.parametrize("signal_name", TREE_WITH_LOAD_SIGNAL_PARAMS)
+@pytest.mark.parametrize("omni_server_function", DIFFUSION_VIDEO_PARAMS, indirect=True)
+def test_reliability_fault_process_kill_tree_with_load_fast_fail_and_cleanup(
+    omni_server_function,
+    openai_client_function,
+    signal_name: str,
+) -> None:
+    """Black-box: during in-flight video requests, kill server process tree and verify post-fault behavior/cleanup."""
+    request_config = _video_request_config()
+    scenario = f"kill_serve_tree_with_load_{signal_name.lower()}"
+    injector = make_server_tree_kill_fault_injector(
+        signal_name=signal_name,
+        post_kill_wait_seconds=2.0,
+        inter_kill_wait_seconds=0.1,
+    )
+    load_result = run_fault_injection_with_rate_load(
+        submit_request=lambda: openai_client_function.send_video_diffusion_request(request_config, request_num=1),
+        inject_fault=lambda: injector(omni_server_function),
+        num_requests=INFLIGHT_INJECTION_REQUEST_COUNT,
+        request_rate=INFLIGHT_INJECTION_REQUEST_RATE,
+        completion_timeout_sec=120.0,
+    )
+    assert load_result["failure_observed"], (
+        f"[{scenario}] expected at least one load request failure after fault; load_result={load_result}"
+    )
+    host = omni_server_function.host
+    port = omni_server_function.port
+    _assert_post_fault_video_fast_fail(host, port, scenario=scenario)
+    _assert_post_fault_health_terminal(host, port, scenario=scenario)
+    assert_no_server_tree_process_residual_and_gpu_release(
+        omni_server_function,
+        scenario=scenario,
+        timeout_sec=worker_residual_timeout_after_kill_signal(signal_name),
+    )