[Bugfix][MiniMax-H3] Use Hopper-safe modulation precision

Signed-off-by: david6666666 <530634352@qq.com>

docs/design/fullduplex-personaplex.md

@@ -271,3 +271,111 @@ The required evidence is:
 
 Audio that is empty, all zero, non-finite, or only a protocol `listen` event is
 not a successful end-to-end result.
+
+### Paced multi-session load driver
+
+The existing driver keeps its two-session admission and slot-reuse validation
+when `--sessions` is omitted. Passing `--sessions N` selects a separate,
+bounded load mode; it does not replace the lifecycle test or change server
+admission capacity. Start an existing PersonaPlex server configured to admit
+the requested number of sessions, then run from a source checkout with the
+vLLM-Omni client dependencies installed:
+
+```bash
+python -m tests.e2e.online_serving.personaplex_realtime_duplex \
+  --model nvidia/personaplex-7b-v1 \
+  --input-wav /path/to/input.wav \
+  --sessions 2 --load-frames 1000 --drain-s 2 \
+  --server-revision SERVER_COMMIT --server-hardware A100-80GB \
+  --output-dir /tmp/personaplex-load-n2
+```
+
+Use a new output directory for every run. The driver refuses to overwrite an
+existing `load-result.json`. The server labels are supplied by the operator,
+not detected or verified by the client. Record the server command, package
+versions, checkpoint revision and hardware alongside a published measurement.
+
+All attempted sessions finish admission before admitted sessions share a
+common input start. A rejected session remains a failed row in the report;
+it is not removed from the denominator. Each session receives at most
+`--load-frames` input frames, without looping the WAV, and uses 80 ms pacing.
+A stalled send cannot cause a catch-up burst. The timeline records send
+schedule lateness so an overloaded client is visible rather than mistaken
+for a faster server. This mode measures steady concurrency, not join/leave
+interference; the original lifecycle mode still checks slot reuse.
+Load mode does not probe an additional N+1 session for `resource_exhausted`;
+the lifecycle mode owns that admission-limit check. A per-session
+`client_pacing_warning` and the summary's `client_pacing_warning_sessions`
+identify send-lateness p99 above one nominal frame (80 ms). This is a warning,
+not an extra acceptance gate: read it alongside server-facing measurements
+so a slow client is not mistaken for a slow server.
+
+The fixed drain window is part of the measurement configuration. Audio
+arriving during cleanup is excluded, so closing cannot rescue a session that
+missed its observation deadline. Cleanup is attempted for admitted sessions
+even when audio validation fails. Protocol errors, missing/invalid audio,
+shared response IDs and cleanup failures make the report fail and the CLI
+exit nonzero. Cancellation closes transports and propagates to the caller.
+`--cleanup-timeout-s` (default 5 s) bounds session-close send and acknowledgement
+together, independently of `--timeout-s`. The subsequent WebSocket close
+handshake has the same separate bound; an unresponsive close acknowledgement
+cannot hold cancellation for the full inference timeout.
+
+`load-result.json` retains per-session send and audio-packet timelines using
+the client's monotonic clock. It does not export handshake credentials or
+raw audio payloads. Metric definitions are explicit:
+
+Both modes validate each nonempty audio packet's strict base64 encoding, PCM16
+sample alignment, 24 kHz sample rate and response identity. Valid aggregate
+audio cannot hide a malformed packet. In load mode, rejected packets make the session fail
+and appear in `invalid_audio_packets` with their event index, receive time and
+fixed diagnostic code; they do not contribute samples or timing intervals.
+Empty audio deltas remain legal and contribute no samples. This check does
+not change the public client's decoding. The lifecycle scenario and dispatch
+remain the same, but malformed packets that aggregate checks once tolerated
+now fail that mode as well.
+
+The probe accepts both the current OpenAI Realtime audio event name
+`response.output_audio.delta` and the legacy `response.audio.delta`. Validation
+uses the wire payload directly rather than relying on a particular client
+library's alias table, so a server-side event-name migration cannot silently
+turn real audio into a zero-output measurement.
+
+| Field | Meaning |
+| --- | --- |
+| `client_audio_packet_interval_ms` | Consecutive nonempty packet arrivals: median, linear p99, max, and zero-based index of the first maximal interval. Interval index 0 ends at packet 1. One packet can contain multiple codec frames; this is not server tick latency. |
+| `client_first_audio_after_stream_start_ms` | First audio receipt minus first input send start; admission is excluded. |
+| `client_stream_rtf` | First input send start to last audio receipt, divided by received audio duration. This includes real-time input pacing and is not an inference-only RTF. |
+| `client_send_lateness_ms` | Send start minus its original shared schedule, clipped at zero. |
+| `frame_deficit` | Integer input-frame minus output-frame count from the shared session validation; missing when valid whole-frame accounting is unavailable. |
+| `output_frames`, `voiced_frames`, `silent_frames`, `frame_coverage_ratio`, `audio_rms`, `audio_chunks` | Session-validation statistics retained on success and on acceptance failure once computed. |
+| `acceptance_check` | Failed local check, with numeric diagnostics in `error`; other exceptions retain phase/type only, without remote handshake or metadata text. |
+| `audio_underrun_s`, `audio_underrun_event_count`, `audio_continuity_ok` | Shared `compute_continuity_stats` on valid PCM16 packet arrivals, with its default 100 ms underrun threshold; absent audio gives missing values rather than a continuity pass. |
+
+No output gives missing latency/RTF values, not zero. By default RTF is
+diagnostic; `--max-client-rtf` adds an explicit ceiling. The existing frame
+coverage, audible-frame and scheduler-data-plane checks still apply. A longer
+`--drain-s` can demonstrate eventual completion under backlog, but it must not
+be reported as real-time capacity when the fixed-window run failed or client
+RTF exceeded the stated target. Likewise, the first live session can pay
+one-time model/codec/JIT costs; either use the server's configured duplex
+warmup path or label that run as cold-start evidence before comparing steady
+capacity. Passing client-only metrics does not certify model quality or a
+real-time capacity target on another GPU. Begin with N=1 and N=2; higher N is a
+capacity experiment, not a promised property of an A100 or of this patch.
+Continuity is a client-arrival diagnostic, not proof of speaker playback and
+not an additional pass/fail gate. Packet-interval and send-lateness summaries
+retain their documented linear p99; the older client interval helper reports
+rounded nearest-rank p95 instead and is not substituted for that schema.
+
+Deterministic driver and localhost WebSocket checks need no model or GPU:
+
+```bash
+python -m pytest tests/e2e/online_serving/test_personaplex_load_driver.py \
+  -m 'core_model and cpu' --run-level core_model -q
+```
+
+These checks test the driver against controlled protocol responses. Run the
+load-driver command against real weights separately before reporting model-serving
+performance. Server tick timestamps and generic TTFT/TPOT aggregation remain
+the responsibility of the existing duplex metrics paths, not this driver.

recipes/MiniMaxAI/MiniMax-H3.md

@@ -1106,6 +1106,68 @@ generations move the end-to-end figure toward the denoising one. Fusing the
 adapter does not measurably change startup: weight loading took 77.3 s with it
 against 85.8 s without.
 
+### FastH3 8-Step V2 full checkpoint
+
+[FastH3 8-Step V2](https://huggingface.co/FastVideo/FastVideo-FastH3-8-Step-V2)
+ships a full Diffusers-format transformer, including learned VSA compression
+gates. Serve the Hugging Face model ID or a downloaded snapshot directly:
+
+```bash
+export FASTH3_V2_MODEL=FastVideo/FastVideo-FastH3-8-Step-V2
+```
+
+Use a Hugging Face account with access to the release, and pin `--revision`
+when reproducing a result. The existing component loader reads its safetensors
+shards; the H3 weight loader maps names and loads Q/K/V and MLP projections into
+native parameters in memory. No conversion command or second transformer
+checkpoint is needed.
+
+The transformer and text components come from the V2 release. To retain Omni's
+native tiled and parallel VAE execution, the loader fetches only the video/audio
+VAEs from the `MiniMaxAI/MiniMax-H3` revision pinned in V2's `provenance.json`.
+Those components use the normal Hugging Face cache. No separate base-model
+argument or full base-transformer download is required. Download time is a
+preparation cost, separate from warmup and request latency.
+
+Install the optional `fastvideo-kernel` build required by the
+`FASTVIDEO_VSA` backend, including its H3 block-map entry point. A four-GPU
+configuration is:
+
+```bash
+CUDA_VISIBLE_DEVICES=0,1,2,3 vllm serve "${FASTH3_V2_MODEL}" --omni \
+  --host 127.0.0.1 --port 8095 --trust-remote-code --task-type fl2va \
+  --served-model-name FastH3-V2 \
+  --num-gpus 4 --usp 4 --ring 1 \
+  --vae-patch-parallel-size 4 --vae-parallel-mode tile --vae-use-tiling \
+  --diffusion-attention-backend FASTVIDEO_VSA
+
+curl --fail-with-body -sS http://127.0.0.1:8095/v1/videos/sync \
+  -F 'model=FastH3-V2' \
+  -F 'prompt=A red fox runs through fresh snow at dawn, with fast pawsteps, winter wind and distant birds.' \
+  -F 'width=1344' -F 'height=768' -F 'aspect_ratio=16:9' -F 'fps=24' \
+  -F 'num_inference_steps=8' -F 'guidance_scale=1' -F 'flow_shift=10' \
+  -F 'seed=1101' \
+  -F 'extra_params={"task":"t2va","duration":4.4,"audio_flow_shift":3.0}' \
+  -o fasth3_v2.mp4
+```
+
+V2 pins video/audio shifts **10/3**, guidance **1**, and the unshifted ladder
+`[0.999, 0.874, 0.749, 0.624, 0.5, 0.375, 0.25, 0.125, 0.0]`.
+Use **`num_inference_steps=8`** in Omni: these nine nodes bound eight model
+evaluations. FastVideo's `--steps 9` counts nodes instead. The existing H3 ODE
+update is reused without stochastic re-noising.
+
+The trained attention policy is **80% sparsity with 64-token video tiles**.
+Omni derives the number of selected video tiles from each request's geometry;
+do not supply a fixed `--fastvideo-vsa-topk`. Text/condition/audio prefix keys
+remain available to every query, and prefix queries remain dense. Missing VSA
+geometry or a failed H3 kernel raises an error instead of serving the student
+with dense attention.
+
+This release supports T2VA only, with local or pure Ulysses attention. Additional
+LoRA adapters, Ref2VA and FL2VA conditioning are unsupported. The legacy
+four-step adapter keeps its original sampling and fixed-top-k behavior.
+
 ## Key parameters
 
 | Parameter | Recommended value | Notes |

tests/diffusion/models/minimax_h3/test_fasth3_checkpoint.py

@@ -0,0 +1,679 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable, Iterable
+
+import pytest
+import torch
+import torch.nn as nn
+
+from vllm_omni.diffusion.data import AttentionConfig, AttentionSpec, DiffusionParallelConfig, OmniDiffusionConfig
+from vllm_omni.diffusion.models.minimax_h3.fasth3_checkpoint import (
+    FASTH3_BASE_MODEL,
+    FASTH3_V2_BASE_SCHEDULE,
+    FASTH3_V2_MODEL_ID,
+    FastH3CheckpointSpec,
+)
+from vllm_omni.diffusion.models.minimax_h3.minimax_h3_transformer import (
+    MiniMaxH3DiTArchConfig,
+    MiniMaxH3DiTModel,
+)
+from vllm_omni.diffusion.models.minimax_h3.pipeline_minimax_h3 import (
+    resolve_minimax_h3_diffusion_model_path,
+)
+from vllm_omni.diffusion.sched.sigma_schedule import DMD2SigmaSchedule
+from vllm_omni.errors import OmniClientError
+from vllm_omni.inputs.data import OmniDiffusionSamplingParams
+from vllm_omni.lora.request import LoRARequest
+from vllm_omni.model_executor.models.minimax_h3.checkpoint import (
+    is_minimax_h3_modular,
+    resolve_minimax_h3_model_root,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu]
+
+
+def _fastvideo_metadata() -> dict[str, object]:
+    """Return the release-owned metadata read from fastvideo_inference.json."""
+    return {
+        "model_id": FASTH3_V2_MODEL_ID,
+        "schema_version": "fasth3-inference-contract-v1",
+        "dmd_denoising_steps": [999, 874, 749, 624, 500, 375, 250, 125],
+        "transformer_forwards": 8,
+        "video_scheduler_shift": 10.0,
+        "audio_scheduler_shift": 3.0,
+        "guidance_scale": 1.0,
+        "attention_backend": "VIDEO_SPARSE_ATTN_H3",
+        "vsa_sparsity": 0.8,
+        "vsa_tile_size": 64,
+        "task": "t2av",
+    }
+
+
+def _sampling(
+    *,
+    num_inference_steps: int | str | None = 8,
+    guidance_scale: float | None = 1.0,
+    extra_args: dict[str, object] | None = None,
+    lora_request: LoRARequest | None = None,
+    timesteps: torch.Tensor | None = None,
+    sigmas: list[float] | None = None,
+) -> OmniDiffusionSamplingParams:
+    return OmniDiffusionSamplingParams(
+        # Keep the string form to exercise malformed client input validation.
+        num_inference_steps=num_inference_steps,  # type: ignore[arg-type]
+        guidance_scale=guidance_scale,
+        extra_args=extra_args or {},
+        lora_request=lora_request,
+        timesteps=timesteps,
+        sigmas=sigmas,
+    )
+
+
+def _od_config(
+    *,
+    backend: str = "FASTVIDEO_VSA",
+    topk: int | None = None,
+    per_role: dict[str, AttentionSpec] | None = None,
+    lora_path: str | None = None,
+    ring_degree: int = 1,
+    allgather_degree: int = 1,
+) -> OmniDiffusionConfig:
+    attention_config = AttentionConfig(
+        default=AttentionSpec(backend=backend, fastvideo_vsa_topk=topk),
+        per_role=per_role or {},
+    )
+    return OmniDiffusionConfig(
+        diffusion_attention_config=attention_config,
+        lora_path=lora_path,
+        parallel_config=DiffusionParallelConfig(ring_degree=ring_degree, allgather_degree=allgather_degree),
+    )
+
+
+def test_v2_release_has_nine_sigma_nodes_and_eight_intervals():
+    assert FASTH3_V2_BASE_SCHEDULE.base_schedule == (
+        0.999,
+        0.874,
+        0.749,
+        0.624,
+        0.5,
+        0.375,
+        0.25,
+        0.125,
+        0.0,
+    )
+    assert len(FASTH3_V2_BASE_SCHEDULE.base_schedule) == 9
+    assert FASTH3_V2_BASE_SCHEDULE.num_inference_steps == 8
+    assert FASTH3_V2_BASE_SCHEDULE.shifted_sigmas(10.0) == pytest.approx(
+        [
+            0.9998999099189271,
+            0.985788405143244,
+            0.9675752486758817,
+            0.94316807738815,
+            0.9090909090909091,
+            0.8571428571428571,
+            0.7692307692307693,
+            0.5882352941176471,
+            0.0,
+        ],
+        abs=1e-12,
+    )
+    assert FASTH3_V2_BASE_SCHEDULE.shifted_sigmas(3.0) == pytest.approx(
+        [
+            0.9996664442961973,
+            0.9541484716157204,
+            0.8995196156925539,
+            0.8327402135231315,
+            0.75,
+            0.6428571428571429,
+            0.5,
+            0.3,
+            0.0,
+        ],
+        abs=1e-12,
+    )
+
+
+def test_fastvideo_metadata_builds_release_sampling_metadata():
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+
+    assert spec.vsa_sparsity == 0.8
+    assert spec.release_metadata() == {
+        "partition": "fl2va",
+        "tasks": ["t2va"],
+        "sigma_shift_scales": {"video": 10.0, "audio": 3.0},
+        "base_schedule": list(FASTH3_V2_BASE_SCHEDULE.base_schedule),
+    }
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("model_id", "FastVideo/FastH3-other"),
+        ("schema_version", "fasth3-inference-contract-v0"),
+        ("dmd_denoising_steps", [999, 749, 500, 250]),
+        ("transformer_forwards", 4),
+        ("video_scheduler_shift", 12.0),
+        ("audio_scheduler_shift", 4.0),
+        ("guidance_scale", 2.0),
+        ("attention_backend", "FLASH_ATTN"),
+        ("vsa_sparsity", 0.9),
+        ("vsa_tile_size", 32),
+        ("task", "t2va"),
+    ],
+)
+def test_fastvideo_metadata_rejects_contract_drift(field, value):
+    metadata = _fastvideo_metadata()
+    metadata[field] = value
+
+    with pytest.raises(ValueError, match=field):
+        FastH3CheckpointSpec.from_metadata(metadata)
+
+
+def test_fastvideo_metadata_rejects_missing_contract_field():
+    metadata = _fastvideo_metadata()
+    metadata.pop("task")
+
+    with pytest.raises(ValueError, match="task"):
+        FastH3CheckpointSpec.from_metadata(metadata)
+
+
+def test_legacy_h3_metadata_is_not_treated_as_fastvideo_release():
+    with pytest.raises(ValueError, match="model_id"):
+        FastH3CheckpointSpec.from_metadata(
+            {
+                "schema_version": 1,
+                "partition": "fl2va",
+                "tasks": ["t2va", "fl2va"],
+                "sigma_shift_scales": {"video": 12.0, "audio": 3.0},
+            }
+        )
+
+
+def test_v2_serving_contract_accepts_backend_selected_for_h3_self_role():
+    config = _od_config(
+        backend="TORCH_SDPA",
+        per_role={"self": AttentionSpec(backend="FASTVIDEO_VSA")},
+    )
+    FastH3CheckpointSpec.from_metadata(_fastvideo_metadata()).check_serving_contract(
+        partition="fl2va", od_config=config
+    )
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "partition", "match"),
+    [
+        ({}, "ref2va", "task-type fl2va"),
+        ({"lora_path": "/tmp/adapter"}, "fl2va", "LoRA"),
+        ({"backend": "TORCH_SDPA"}, "fl2va", "FASTVIDEO_VSA"),
+        ({"topk": 2}, "fl2va", "fixed fastvideo_vsa_topk"),
+        ({"ring_degree": 2}, "fl2va", "local attention"),
+        ({"allgather_degree": 2}, "fl2va", "local attention"),
+    ],
+)
+def test_v2_serving_contract_rejects_incompatible_runtime(kwargs, partition, match):
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+    with pytest.raises(ValueError, match=match):
+        spec.check_serving_contract(partition=partition, od_config=_od_config(**kwargs))
+
+
+def test_v2_serving_contract_reads_a_per_role_fixed_topk_override():
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+    config = _od_config(
+        per_role={"self": AttentionSpec(backend="FASTVIDEO_VSA", fastvideo_vsa_topk=1)},
+    )
+    with pytest.raises(ValueError, match="fixed fastvideo_vsa_topk"):
+        spec.check_serving_contract(partition="fl2va", od_config=config)
+
+
+def test_v2_request_contract_accepts_omitted_steps_and_default_shifts():
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+    spec.check_request(_sampling(num_inference_steps=None))
+    spec.check_request(_sampling(extra_args={"flow_shift": 10, "audio_flow_shift": 3}, guidance_scale=1))
+
+
+@pytest.mark.parametrize(
+    "sampling",
+    [
+        _sampling(num_inference_steps=None),
+        _sampling(timesteps=torch.tensor([0.9, 0.5, 0.0])),
+        _sampling(sigmas=[1.0, 0.5, 0.0]),
+    ],
+)
+def test_v2_step_execution_requires_pinned_count_and_schedule(sampling):
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+    with pytest.raises(OmniClientError, match="step execution"):
+        spec.check_request(sampling, step_execution=True)
+
+
+def test_v2_step_execution_accepts_only_explicit_pinned_count():
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+    spec.check_request(_sampling(num_inference_steps=8), step_execution=True)
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "match"),
+    [
+        ({"num_inference_steps": 7}, "num_inference_steps=8"),
+        ({"num_inference_steps": 9}, "num_inference_steps=8"),
+        ({"num_inference_steps": "8"}, "num_inference_steps=8"),
+        ({"extra_args": {"flow_shift": 9}}, "flow_shift=10"),
+        ({"extra_args": {"audio_flow_shift": 4}}, "audio_flow_shift=3"),
+        ({"extra_args": {"flow_shift": "bad"}}, "flow_shift=10"),
+        ({"guidance_scale": 2.0}, "guidance_scale=1"),
+        (
+            {"lora_request": LoRARequest(lora_name="test", lora_int_id=1, lora_path="/tmp/test.safetensors")},
+            "per-request LoRA",
+        ),
+    ],
+)
+def test_v2_request_contract_rejects_sampling_drift(kwargs, match):
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+    with pytest.raises(OmniClientError, match=match):
+        spec.check_request(_sampling(**kwargs))
+
+
+def test_resolve_native_vaes_uses_pinned_revision_and_only_vae_patterns(tmp_path, monkeypatch):
+    model_root = tmp_path / "FastH3"
+    model_root.mkdir()
+    revision = "base-revision-123"
+    (model_root / "provenance.json").write_text(
+        json.dumps({"base_model": f"hf://{FASTH3_BASE_MODEL}@{revision}"}),
+        encoding="utf-8",
+    )
+    snapshot = tmp_path / "hf-snapshot"
+    calls: dict[str, object] = {}
+
+    def fake_download(**kwargs: object) -> str:
+        calls.update(kwargs)
+        snapshot.mkdir()
+        return str(snapshot)
+
+    monkeypatch.setattr(
+        "vllm_omni.diffusion.models.minimax_h3.fasth3_checkpoint.download_weights_from_hf_specific",
+        fake_download,
+    )
+
+    result = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata()).resolve_native_vaes(model_root)
+
+    assert result == snapshot / "FL2VA"
+    assert calls == {
+        "model_name_or_path": FASTH3_BASE_MODEL,
+        "cache_dir": None,
+        "allow_patterns": ["FL2VA/video_vae/**", "FL2VA/audio_vae/**"],
+        "revision": revision,
+        "require_all": True,
+    }
+    assert sorted(path.name for path in model_root.iterdir()) == ["provenance.json"]
+    assert not (model_root / "model_index.json").exists()
+    assert not (model_root / "conversion.json").exists()
+
+
+@pytest.mark.parametrize(
+    "base_model",
+    [
+        "MiniMaxAI/MiniMax-H3@main",
+        "hf://MiniMaxAI/MiniMax-H3@",
+        "hf://Other/MiniMax-H3@revision",
+        "",
+        None,
+    ],
+)
+def test_resolve_native_vaes_rejects_unpinned_provenance(tmp_path, base_model):
+    model_root = tmp_path / "FastH3"
+    model_root.mkdir()
+    (model_root / "provenance.json").write_text(json.dumps({"base_model": base_model}), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="provenance"):
+        FastH3CheckpointSpec.from_metadata(_fastvideo_metadata()).resolve_native_vaes(model_root)
+
+
+def test_modular_local_model_root_uses_text_encoder_without_partition_files(tmp_path):
+    root = tmp_path / "FastH3"
+    (root / "text_encoder").mkdir(parents=True)
+    (root / "text_encoder" / "config.json").write_text("{}", encoding="utf-8")
+    (root / "modular_model_index.json").write_text(
+        json.dumps({"_class_name": "MiniMaxH3ModularPipeline"}),
+        encoding="utf-8",
+    )
+
+    assert is_minimax_h3_modular(str(root)) is True
+    assert resolve_minimax_h3_model_root(str(root), None, "fl2va") == str(root / "text_encoder")
+    assert not (root / "FL2VA").exists()
+
+
+def test_modular_hub_model_root_downloads_only_text_encoder(monkeypatch, tmp_path):
+    from vllm_omni.model_executor.models.minimax_h3 import checkpoint as checkpoint_module
+
+    snapshot = tmp_path / "modular-snapshot"
+    calls: dict[str, object] = {}
+
+    def fake_index(model: str, *, revision: str | None = None) -> dict[str, str]:
+        assert model == "FastVideo/FastVideo-FastH3-8-Step-V2"
+        assert revision == "release-revision"
+        return {"_class_name": "MiniMaxH3ModularPipeline"}
+
+    def fake_download(**kwargs: object) -> str:
+        calls.update(kwargs)
+        snapshot.mkdir()
+        return str(snapshot)
+
+    monkeypatch.setattr(checkpoint_module, "get_diffusion_model_index", fake_index)
+    monkeypatch.setattr(checkpoint_module, "download_weights_from_hf_specific", fake_download)
+
+    result = resolve_minimax_h3_model_root(
+        "FastVideo/FastVideo-FastH3-8-Step-V2",
+        "release-revision",
+        "fl2va",
+    )
+
+    assert result == str(snapshot / "text_encoder")
+    assert calls == {
+        "model_name_or_path": "FastVideo/FastVideo-FastH3-8-Step-V2",
+        "cache_dir": None,
+        "allow_patterns": ["text_encoder/**"],
+        "revision": "release-revision",
+        "require_all": True,
+    }
+
+
+def test_native_hub_model_with_modular_class_keeps_partition_layout(monkeypatch):
+    from vllm_omni.model_executor.models.minimax_h3 import checkpoint as checkpoint_module
+
+    monkeypatch.setattr(
+        checkpoint_module,
+        "get_diffusion_model_index",
+        lambda _model, *, revision=None: {"_class_name": "MiniMaxH3ModularPipeline"},
+    )
+
+    assert checkpoint_module.is_minimax_h3_modular("MiniMaxAI/MiniMax-H3") is False
+
+
+def test_native_local_model_root_keeps_partition_layout(tmp_path):
+    root = tmp_path / "MiniMax-H3"
+    (root / "FL2VA" / "text_encoder").mkdir(parents=True)
+    (root / "FL2VA" / "text_encoder" / "config.json").write_text("{}", encoding="utf-8")
+
+    assert is_minimax_h3_modular(str(root)) is False
+    assert resolve_minimax_h3_model_root(str(root), None, "fl2va") == str(root / "FL2VA" / "text_encoder")
+
+
+def test_diffusion_resolver_returns_local_modular_root_without_partition_files(tmp_path):
+    root = tmp_path / "FastH3"
+    root.mkdir()
+    (root / "modular_model_index.json").write_text(
+        json.dumps({"_class_name": "MiniMaxH3ModularPipeline"}),
+        encoding="utf-8",
+    )
+
+    assert resolve_minimax_h3_diffusion_model_path(str(root), None, "fl2va") == str(root)
+    assert not (root / "FL2VA").exists()
+    assert not (root / "Ref2VA").exists()
+
+
+def test_diffusion_resolver_hub_modular_downloads_transformer_contract_only(monkeypatch, tmp_path):
+    from vllm_omni.diffusion.models.minimax_h3 import pipeline_minimax_h3 as pipeline_module
+
+    snapshot = tmp_path / "modular-snapshot"
+    calls: dict[str, object] = {}
+
+    def fake_modular(_model: str, _revision: str | None = None) -> bool:
+        return True
+
+    def fake_download(**kwargs: object) -> str:
+        calls.update(kwargs)
+        snapshot.mkdir()
+        return str(snapshot)
+
+    monkeypatch.setattr(pipeline_module, "is_minimax_h3_modular", fake_modular)
+    monkeypatch.setattr(pipeline_module, "download_weights_from_hf_specific", fake_download)
+
+    result = resolve_minimax_h3_diffusion_model_path(
+        "FastVideo/FastVideo-FastH3-8-Step-V2", "release-revision", "fl2va"
+    )
+
+    assert result == str(snapshot)
+    assert calls == {
+        "model_name_or_path": "FastVideo/FastVideo-FastH3-8-Step-V2",
+        "cache_dir": None,
+        "allow_patterns": [
+            "modular_model_index.json",
+            "fastvideo_inference.json",
+            "provenance.json",
+            "transformer/**",
+        ],
+        "revision": "release-revision",
+        "require_all": True,
+    }
+    patterns = calls["allow_patterns"]
+    assert isinstance(patterns, list)
+    assert all("video_vae" not in pattern and "audio_vae" not in pattern for pattern in patterns)
+
+
+def test_diffusers_arch_config_normalizes_all_aliases():
+    arch = MiniMaxH3DiTArchConfig.from_mapping(
+        {
+            "num_refiner_layers": 3,
+            "ffn_dim": 12,
+            "in_channels": 4,
+            "audio_in_channels": 6,
+            "freq_dim": 8,
+            "time_embed_hidden_dim": 10,
+            "rope_freq_dim": 5,
+            "patch_size": [1, 2, 2],
+        }
+    )
+
+    assert arch.token_refiner_num_layers == 3
+    assert arch.ffn_hidden_size == 12
+    assert arch.latents_dim == 4
+    assert arch.audio_latents_dim == 6
+    assert arch.timestep_input_dim == 8
+    assert arch.time_embed_hidden_size == 10
+    assert arch.rope_inv_freq_len == 5
+    assert arch.patch_size == (1, 2, 2)
+
+
+class _WeightTarget(nn.Module):
+    def __init__(self, shape: tuple[int, ...], loader: Callable[..., None]) -> None:
+        super().__init__()
+        self.weight = nn.Parameter(torch.zeros(shape))
+        self.weight.weight_loader = loader
+
+
+class _PipelineLoadTransformer(nn.Module):
+    def __init__(self, loaded: set[str]) -> None:
+        super().__init__()
+        self.arch = MiniMaxH3DiTArchConfig(num_layers=1)
+        self._loaded = loaded
+
+    def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
+        list(weights)
+        return self._loaded
+
+    def post_load_weights(self) -> None:
+        pass
+
+
+def test_pipeline_load_weights_rejects_missing_fast_h3_compression_gate():
+    from vllm_omni.diffusion.models.minimax_h3.pipeline_minimax_h3 import MiniMaxH3Pipeline
+
+    pipeline = object.__new__(MiniMaxH3Pipeline)
+    nn.Module.__init__(pipeline)
+    pipeline.transformer = _PipelineLoadTransformer({"blocks.0.attn.qkv_proj.weight"})
+    pipeline.video_vae = None
+    pipeline.audio_vae = None
+    pipeline._fasth3 = None
+    pipeline._fasth3_checkpoint = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+
+    with pytest.raises(ValueError, match="missing compression gates"):
+        pipeline.load_weights([("transformer.blocks.0.attn.qkv_proj.weight", torch.zeros(1))])
+
+
+def _load_fixture(
+    *, diffusers_weights: bool
+) -> tuple[MiniMaxH3DiTModel, list[tuple[str | None, torch.Tensor]], list[tuple[int, torch.Tensor]]]:
+    model = object.__new__(MiniMaxH3DiTModel)
+    nn.Module.__init__(model)
+    model.arch = MiniMaxH3DiTArchConfig(
+        num_layers=1,
+        hidden_size=3,
+        num_attention_heads=2,
+        attention_head_dim=2,
+        ffn_hidden_size=6,
+        rope_inv_freq_len=2,
+    )
+    model._diffusers_weights = diffusers_weights
+    model._rope_theta = 10000.0
+    qkv_calls: list[tuple[str | None, torch.Tensor]] = []
+    fc1_calls: list[tuple[int, torch.Tensor]] = []
+
+    def qkv_loader(param: nn.Parameter, loaded_weight: torch.Tensor, shard_id: str | None = None) -> None:
+        if shard_id is None:
+            assert loaded_weight.shape == (12, 3)
+            param.data.copy_(loaded_weight)
+            qkv_calls.append((None, loaded_weight.clone()))
+            return
+        assert loaded_weight.shape == (4, 3)
+        assert shard_id in {"q", "k", "v"}
+        offset = {"q": 0, "k": 4, "v": 8}[shard_id]
+        param.data[offset : offset + 4].copy_(loaded_weight)
+        qkv_calls.append((shard_id, loaded_weight.clone()))
+
+    def fc1_loader(param: nn.Parameter, loaded_weight: torch.Tensor, shard_id: int) -> None:
+        assert loaded_weight.shape == (3, 3)
+        param.data[shard_id * 3 : (shard_id + 1) * 3].copy_(loaded_weight)
+        fc1_calls.append((shard_id, loaded_weight.clone()))
+
+    model.blocks = nn.ModuleList([nn.Module()])
+    model.blocks[0].attn = nn.Module()
+    model.blocks[0].attn.qkv_proj = _WeightTarget((12, 3), qkv_loader)
+    model.blocks[0].mlp = nn.Module()
+    model.blocks[0].mlp.fc1 = _WeightTarget((6, 3), fc1_loader)
+    model.rope = nn.Module()
+    model.rope.register_buffer("inv_freq", torch.zeros(2))
+    return model, qkv_calls, fc1_calls
+
+
+def test_diffusers_load_weights_uses_shape_aware_qkv_and_mlp_slices():
+    model, qkv_calls, fc1_calls = _load_fixture(diffusers_weights=True)
+    q = torch.arange(12, dtype=torch.float32).reshape(4, 3) + 10
+    k = torch.arange(12, dtype=torch.float32).reshape(4, 3) + 100
+    v = torch.arange(12, dtype=torch.float32).reshape(4, 3) + 200
+    mlp = torch.arange(18, dtype=torch.float32).reshape(6, 3) + 300
+
+    loaded = model.load_weights(
+        [
+            ("transformer_blocks.0.attn.to_q.weight", q),
+            ("transformer_blocks.0.attn.to_k.weight", k),
+            ("transformer_blocks.0.attn.to_v.weight", v),
+            ("transformer_blocks.0.ff.net.0.proj.weight", mlp),
+        ]
+    )
+
+    assert loaded == {
+        "blocks.0.attn.qkv_proj.weight",
+        "blocks.0.mlp.fc1.weight",
+        "rope.inv_freq",
+    }
+    assert [shard_id for shard_id, _ in qkv_calls] == ["q", "k", "v"]
+    assert [shard_id for shard_id, _ in fc1_calls] == [0, 1]
+    torch.testing.assert_close(model.blocks[0].attn.qkv_proj.weight, torch.cat((q, k, v)))
+    torch.testing.assert_close(model.blocks[0].mlp.fc1.weight, torch.cat((mlp[3:], mlp[:3])))
+    torch.testing.assert_close(
+        model.rope.inv_freq,
+        1.0 / (10000.0 ** (torch.arange(0, 4, 2, dtype=torch.float32) / 4)),
+        rtol=0,
+        atol=0,
+    )
+
+
+def test_native_load_weights_keeps_grouped_qkv_and_gate_first_layouts():
+    model, qkv_calls, fc1_calls = _load_fixture(diffusers_weights=False)
+    grouped_qkv = torch.arange(36, dtype=torch.float32).reshape(12, 3)
+    fc1 = torch.arange(18, dtype=torch.float32).reshape(6, 3)
+
+    loaded = model.load_weights(
+        [
+            ("blocks.0.attn.qkv_proj.weight", grouped_qkv),
+            ("blocks.0.mlp.fc1.weight", fc1),
+        ]
+    )
+
+    expected_qkv = torch.cat(
+        (
+            grouped_qkv[:2],
+            grouped_qkv[6:8],
+            grouped_qkv[2:4],
+            grouped_qkv[8:10],
+            grouped_qkv[4:6],
+            grouped_qkv[10:],
+        )
+    )
+    assert loaded == {"blocks.0.attn.qkv_proj.weight", "blocks.0.mlp.fc1.weight"}
+    assert [shard_id for shard_id, _ in qkv_calls] == [None]
+    assert [shard_id for shard_id, _ in fc1_calls] == [0, 1]
+    torch.testing.assert_close(model.blocks[0].attn.qkv_proj.weight, expected_qkv)
+    torch.testing.assert_close(model.blocks[0].mlp.fc1.weight, fc1)
+    assert torch.equal(model.rope.inv_freq, torch.zeros(2))
+
+
+def test_diffusers_load_weights_rejects_unknown_name():
+    model, _, _ = _load_fixture(diffusers_weights=True)
+
+    with pytest.raises(ValueError, match="unsupported Diffusers H3 weight"):
+        model.load_weights([("transformer_blocks.0.attn.not_a_projection.weight", torch.zeros(4, 3))])
+
+
+def test_diffusers_load_weights_rejects_duplicate_source_name():
+    model, _, _ = _load_fixture(diffusers_weights=True)
+    weight = torch.zeros(4, 3)
+
+    with pytest.raises(ValueError, match="duplicate Diffusers H3 weight"):
+        model.load_weights(
+            [
+                ("transformer_blocks.0.attn.to_q.weight", weight),
+                ("transformer_blocks.0.attn.to_q.weight", weight),
+            ]
+        )
+
+
+def test_diffusers_load_weights_rejects_incomplete_qkv_group():
+    model, _, _ = _load_fixture(diffusers_weights=True)
+
+    with pytest.raises(ValueError, match="incomplete Diffusers H3 QKV"):
+        model.load_weights(
+            [
+                ("transformer_blocks.0.attn.to_q.weight", torch.zeros(4, 3)),
+                ("transformer_blocks.0.attn.to_k.weight", torch.zeros(4, 3)),
+            ]
+        )
+
+
+def test_diffusers_load_weights_rejects_missing_compression_gate_parameter():
+    model, _, _ = _load_fixture(diffusers_weights=True)
+
+    with pytest.raises(ValueError, match="has no model parameter"):
+        model.load_weights([("transformer_blocks.0.attn.to_gate_compress.weight", torch.zeros(4, 3))])
+
+
+def test_resolve_sigma_positions_uses_release_metadata_schedule():
+    from vllm_omni.diffusion.models.minimax_h3 import MiniMaxH3Pipeline
+
+    pipeline = object.__new__(MiniMaxH3Pipeline)
+    nn.Module.__init__(pipeline)
+    pipeline._fasth3 = None
+    pipeline._lora_sigma_schedules = {}
+    spec = FastH3CheckpointSpec.from_metadata(_fastvideo_metadata())
+    pipeline._base_schedule_by_partition = {
+        "fl2va": DMD2SigmaSchedule.from_metadata(spec.release_metadata()),
+    }
+
+    positions, steps = pipeline._resolve_sigma_positions("t2va", _sampling())
+    assert positions == FASTH3_V2_BASE_SCHEDULE.base_schedule
+    assert steps == 8 == len(positions) - 1
+    with pytest.raises(OmniClientError, match="must be 8"):
+        pipeline._resolve_sigma_positions("t2va", _sampling(num_inference_steps=9))

tests/diffusion/models/minimax_h3/test_minimax_h3_contract.py

@@ -423,6 +423,7 @@ def test_pipeline_loads_task_selected_components_with_encoder_ownership(
     assert pipeline.partition == expected_partition
     assert pipeline.supported_tasks == expected_tasks
     assert len(created["dit"]) == expected_dits
+    assert all(kwargs["diffusers_weights"] is False for _, kwargs in created["dit"])
     component_path = tmp_path / component_partition
     assert created["video_vae"] == [str(component_path / "video_vae")]
     assert created["audio_vae"] == [str(component_path / "audio_vae")]

tests/diffusion/models/minimax_h3/test_minimax_h3_fasth3.py

@@ -262,7 +262,10 @@ def test_a_vsa_variant_is_recognised_by_its_compression_gates(tmp_path):
 
 def test_a_tensor_naming_no_h3_parameter_is_an_error(tmp_path):
     path = tmp_path / "fasth3" / "adapter_model.safetensors"
-    _write_adapter(path, tensors={"transformer_blocks.0.attn.norm_q.lora_A.weight": torch.ones((_RANK, _HIDDEN))})
+    _write_adapter(
+        path,
+        tensors={"transformer_blocks.0.attn.not_a_projection.lora_A.weight": torch.ones((_RANK, _HIDDEN))},
+    )
     # Dropping it silently would load a model that is not the distilled student.
     with pytest.raises(FastH3AdapterError, match="name no known"):
         _load(path.parent)

tests/diffusion/models/minimax_h3/test_minimax_h3_modulation.py

@@ -10,6 +10,7 @@ from vllm_omni.diffusion.layers.indexed_modulation import (
     _MAX_1D_GRID_SIZE,
     _iter_row_chunks,
     _launch_row_chunks,
+    _use_hopper_bf16_affine_semantics,
     indexed_gate,
     indexed_gate_rms_norm_scale_shift,
     indexed_scale_shift_,
@@ -190,3 +191,55 @@ def test_fused_modulation_preserves_bf16_residual_boundary() -> None:
     )
 
     assert torch.equal(modulated_out, expected)
+
+
+@pytest.mark.cuda
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+@pytest.mark.skipif(
+    torch.cuda.is_available() and torch.cuda.get_device_capability()[0] != 9,
+    reason="SM90-specific precision path",
+)
+@pytest.mark.skipif(not HAS_TRITON, reason="Triton required")
+def test_hopper_fused_modulation_matches_pytorch_reference() -> None:
+    torch.manual_seed(42)
+    rows, hidden_size, conditions = 128, 3072, 4
+    residual = torch.randn(rows, hidden_size, device="cuda", dtype=torch.bfloat16)
+    gate = torch.randn(conditions, hidden_size, device="cuda", dtype=torch.bfloat16)
+    branch = torch.randn(rows, hidden_size, device="cuda", dtype=torch.bfloat16)
+    weight = torch.randn(hidden_size, device="cuda", dtype=torch.bfloat16)
+    shift = torch.randn(conditions, hidden_size, device="cuda", dtype=torch.bfloat16)
+    scale = torch.randn(conditions, hidden_size, device="cuda", dtype=torch.bfloat16)
+    indices = torch.arange(rows, device="cuda") % conditions
+    eps = 1e-6
+
+    reference_residual = (residual + gate.index_select(0, indices) * branch).to(torch.bfloat16)
+    normalized = reference_residual.float()
+    variance = normalized.pow(2).mean(-1, keepdim=True)
+    normalized = (weight.float() * (normalized * torch.rsqrt(variance + eps))).to(torch.bfloat16)
+    reference_modulated = (
+        normalized * (1.0 + scale.index_select(0, indices)) + shift.index_select(0, indices)
+    ).to(torch.bfloat16)
+
+    actual_residual, actual_modulated = indexed_gate_rms_norm_scale_shift(
+        residual,
+        gate,
+        branch,
+        weight,
+        shift,
+        scale,
+        indices,
+        eps,
+    )
+
+    torch.testing.assert_close(actual_residual, reference_residual, atol=5e-2, rtol=5e-2)
+    torch.testing.assert_close(actual_modulated, reference_modulated, atol=5e-2, rtol=5e-2)
+
+
+@pytest.mark.cuda
+@pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+def test_hopper_precision_dispatch_is_architecture_scoped(monkeypatch) -> None:
+    tensor = torch.empty(0, device="cuda")
+    monkeypatch.setattr(torch.cuda, "get_device_capability", lambda _device: (9, 0))
+    assert _use_hopper_bf16_affine_semantics(tensor)
+    monkeypatch.setattr(torch.cuda, "get_device_capability", lambda _device: (10, 3))
+    assert not _use_hopper_bf16_affine_semantics(tensor)

tests/diffusion/models/minimax_h3/test_minimax_h3_quantization.py

@@ -1,4 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 
 from types import SimpleNamespace
 from unittest.mock import Mock
@@ -41,7 +42,7 @@ class _FakeAttention(nn.Module):
         super().__init__()
 
 
-def _small_od_config():
+def _small_od_config(*, class_name=None):
     arch = {
         "num_layers": 1,
         "token_refiner_num_layers": 1,
@@ -60,6 +61,8 @@ def _small_od_config():
         "final_adaln_out_features": 2 * 8,
         "rope_inv_freq_len": 2,
     }
+    if class_name is not None:
+        arch["_class_name"] = class_name
     return SimpleNamespace(
         tf_model_config=arch,
         parallel_config=SimpleNamespace(ulysses_degree=1),
@@ -133,6 +136,24 @@ def test_fp8_scope_and_prefix_propagation(monkeypatch):
     )
 
 
+def test_explicit_native_weight_format_overrides_diffusers_class_name(monkeypatch):
+    from vllm_omni.diffusion.models.minimax_h3 import minimax_h3_transformer as h3
+
+    monkeypatch.setattr(h3, "ColumnParallelLinear", _FakeLinear)
+    monkeypatch.setattr(h3, "MergedColumnParallelLinear", _FakeLinear)
+    monkeypatch.setattr(h3, "QKVParallelLinear", _FakeLinear)
+    monkeypatch.setattr(h3, "RowParallelLinear", _FakeLinear)
+    monkeypatch.setattr(h3, "Attention", _FakeAttention)
+    monkeypatch.setattr(h3, "get_tensor_model_parallel_world_size", lambda: 1)
+
+    model = h3.MiniMaxH3DiTModel(
+        _small_od_config(class_name="MiniMaxH3Transformer3DModel"),
+        diffusers_weights=False,
+    )
+
+    assert model._diffusers_weights is False
+
+
 class _WeightTarget(nn.Module):
     def __init__(self, loader):
         super().__init__()

tests/diffusion/models/minimax_h3/test_minimax_h3_quantization_quality.py

@@ -187,7 +187,11 @@ def test_minimax_h3_quantization_quality(config: QualityTestConfig):
         "model": model_ref,
         "enforce_eager": True,
         # The fused BF16 baseline only fits on H100-80GB with encoder TP.
-        "stage_0_tensor_parallel_size": 2,
+        # This test loads the direct FL2VA diffusion pipeline (not the
+        # disaggregated two-stage MiniMax-H3 pipeline), so configure both the
+        # DiT world and the Qwen3-VL encoder TP explicitly.
+        "tensor_parallel_size": 2,
+        "text_encoder_tp_size": 2,
         "vae_use_tiling": True,
         # MiniMax H3's VAE is implemented by code shipped with the checkpoint.
         "trust_remote_code": True,

tests/e2e/offline_inference/test_wan22_t2v.py

@@ -6,6 +6,7 @@ import pytest
 from tests.helpers.mark import hardware_test
 from tests.helpers.runtime import OfflineOmniClient
 from vllm_omni.config.omni_config import VllmOmniDiffusionStageConfig
+from vllm_omni.diffusion.data import OmniDiffusionConfig
 from vllm_omni.inputs.data import OmniDiffusionSamplingParams
 
 MODEL = "Wan-AI/Wan2.2-T2V-A14B-Diffusers"
@@ -21,17 +22,19 @@ def test_structured_diffusion_config_reaches_runtime(omni_runner) -> None:
     """The default WAN resolver launches the typed diffusion config itself."""
     omni = omni_runner.omni
     resolved_stages = omni.engine.stage_configs
+    stage_clients = omni.engine.stage_clients
     assert len(resolved_stages) == 1
+    assert len(stage_clients) == 1
     stage_config = resolved_stages[0]
+    od_config = stage_clients[0].od_config
 
     assert isinstance(stage_config, VllmOmniDiffusionStageConfig)
+    assert isinstance(od_config, OmniDiffusionConfig)
     assert stage_config.stage_id == 0
     assert stage_config.model_stage == "dit"
     assert stage_config.final_output_type == "video"
 
     assert omni.engine.stage_vllm_configs == [None]
-    od_config = omni.get_diffusion_od_config()
-    assert od_config is not None
     assert od_config.stage_id == 0
     assert od_config.model == MODEL
     assert od_config.model_class_name == "WanPipeline"

tests/e2e/online_serving/personaplex_realtime_duplex.py

@@ -10,24 +10,24 @@ import base64
 import hashlib
 import json
 import math
+import time
 import uuid
 import wave
-from collections.abc import Sequence
+from collections.abc import Awaitable, Callable, Sequence
 from pathlib import Path
+from typing import Protocol
 from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
 import numpy as np
 
 try:
     import websockets
-    from websockets.exceptions import ConnectionClosed
+    from websockets.exceptions import ConnectionClosed, WebSocketException
 except ImportError as exc:  # pragma: no cover - driver dependency.
     raise SystemExit("Install websockets first: pip install websockets") from exc
 
-from vllm_omni.clients.duplex import (
-    EventCollector,
-    write_pcm16_wav,
-)
+from vllm_omni.benchmarks.audio_continuity import compute_continuity_stats
+from vllm_omni.clients.duplex import EventCollector, write_pcm16_wav
 from vllm_omni.clients.duplex import (
     wait_for_condition as wait_for,
 )
@@ -35,6 +35,24 @@ from vllm_omni.clients.duplex import (
 SAMPLE_RATE_HZ = 24_000
 FRAME_SAMPLES = 1_920
 FRAME_PERIOD_S = FRAME_SAMPLES / SAMPLE_RATE_HZ
+AUDIO_DELTA_EVENT_TYPES = frozenset({"response.audio.delta", "response.output_audio.delta"})
+
+
+class _AudioAcceptanceError(AssertionError):
+    """A local acceptance failure with numeric diagnostics safe to report."""
+
+    def __init__(self, check: str, message: str, stats: dict[str, object]) -> None:
+        super().__init__(message)
+        self.check = check
+        self.stats = dict(stats)
+
+
+class _ProbeTransport(Protocol):
+    async def recv(self) -> str | bytes: ...
+
+    async def send(self, message: str) -> None: ...
+
+    async def close(self) -> None: ...
 
 
 class RawRealtimeProbe:
@@ -46,29 +64,33 @@ class RawRealtimeProbe:
     keeps a transport-only client and reads everything from ``.events``.
     """
 
-    def __init__(self, url: str, *, max_size: int = 64 * 1024 * 1024) -> None:
+    def __init__(self, url: str, *, max_size: int = 64 * 1024 * 1024, close_timeout_s: float = 10.0) -> None:
         self.url = url
         self.max_size = max_size
+        self.close_timeout_s = close_timeout_s
         self.events = EventCollector()
-        self._ws = None
+        self._ws: _ProbeTransport | None = None
         self._reader_task: asyncio.Task[None] | None = None
 
     async def __aenter__(self) -> RawRealtimeProbe:
-        self._ws = await websockets.connect(self.url, max_size=self.max_size)
+        self._ws = await websockets.connect(self.url, max_size=self.max_size, close_timeout=self.close_timeout_s)
         self._reader_task = asyncio.create_task(self._read_events())
         return self
 
     async def __aexit__(self, exc_type, exc, traceback) -> None:
-        if self._ws is not None:
-            await self._ws.close()
-        if self._reader_task is not None:
-            self._reader_task.cancel()
-            try:
-                await self._reader_task
-            except asyncio.CancelledError:
-                pass
+        try:
+            if self._ws is not None:
+                await self._ws.close()
+        finally:
+            if self._reader_task is not None:
+                self._reader_task.cancel()
+                try:
+                    await self._reader_task
+                except asyncio.CancelledError:
+                    pass
 
     async def _read_events(self) -> None:
+        assert self._ws is not None
         try:
             while True:
                 raw = await self._ws.recv()
@@ -81,6 +103,7 @@ class RawRealtimeProbe:
             return
 
     async def send(self, event: dict[str, object]) -> None:
+        assert self._ws is not None
         await self._ws.send(json.dumps(event))
 
 
@@ -142,6 +165,38 @@ def _events(client: RawRealtimeProbe, event_type: str) -> list[dict[str, object]
     return [event for event in client.events.events if event.get("type") == event_type]
 
 
+def _audio_events(client: RawRealtimeProbe) -> list[dict[str, object]]:
+    """Return audio deltas from both current and legacy Realtime event names."""
+    return [event for event in client.events.events if event.get("type") in AUDIO_DELTA_EVENT_TYPES]
+
+
+def _validated_audio_event(event: dict[str, object]) -> bytes:
+    """Decode one wire audio delta without depending on client event aliases."""
+    encoded = event.get("delta")
+    if not isinstance(encoded, str):
+        raise ValueError("missing_audio_delta")
+    try:
+        chunk = base64.b64decode(encoded, validate=True)
+    except ValueError:
+        raise ValueError("invalid_audio_base64") from None
+    if chunk:
+        if len(chunk) % 2:
+            raise ValueError("unaligned_pcm16_packet")
+        if event.get("sample_rate_hz") != SAMPLE_RATE_HZ:
+            raise ValueError("invalid_audio_sample_rate")
+        if not isinstance(event.get("response_id"), str) or not event.get("response_id"):
+            raise ValueError("missing_audio_response_id")
+    return chunk
+
+
+def _validated_audio_chunks(client: RawRealtimeProbe) -> list[tuple[dict[str, object], bytes]]:
+    return [(event, _validated_audio_event(event)) for event in _audio_events(client)]
+
+
+def _audio_bytes(client: RawRealtimeProbe) -> bytes:
+    return b"".join(chunk for _, chunk in _validated_audio_chunks(client) if chunk)
+
+
 async def _open_session(
     args: argparse.Namespace,
     *,
@@ -151,22 +206,7 @@ async def _open_session(
 ) -> tuple[RawRealtimeProbe, dict[str, object]]:
     client = RawRealtimeProbe(_realtime_url(args.url, args.model, session_id))
     await client.__aenter__()
-    await client.send(
-        {
-            "type": "session.update",
-            "session": {
-                "session_id": session_id,
-                "model": args.model,
-                "modalities": ["audio", "text"],
-                "input_audio_format": "pcm_f32le",
-                "output_audio_format": "pcm16",
-                "voice": args.voice,
-                "instructions": persona,
-                "turn_detection": None,
-                "extra_body": {"auto_response": True},
-            },
-        }
-    )
+    await client.send(_session_update(args, session_id=session_id, persona=persona))
     event_type = "error" if expect_error else "session.created"
     await wait_for(
         lambda: client.events.count(event_type) > 0,
@@ -176,6 +216,23 @@ async def _open_session(
     return client, _events(client, event_type)[-1]
 
 
+def _session_update(args: argparse.Namespace, *, session_id: str, persona: str) -> dict[str, object]:
+    return {
+        "type": "session.update",
+        "session": {
+            "session_id": session_id,
+            "model": args.model,
+            "modalities": ["audio", "text"],
+            "input_audio_format": "pcm_f32le",
+            "output_audio_format": "pcm16",
+            "voice": args.voice,
+            "instructions": persona,
+            "turn_detection": None,
+            "extra_body": {"auto_response": True},
+        },
+    }
+
+
 async def _close_session(client: RawRealtimeProbe, *, timeout_s: float) -> None:
     await client.send({"type": "session.close"})
     await wait_for(
@@ -200,23 +257,24 @@ async def _stream_frames(
             frame_count = min(frame_count, max_frames)
     for seq in range(frame_count):
         frame = pcm[seq * FRAME_SAMPLES : (seq + 1) * FRAME_SAMPLES]
-        frame = np.pad(frame, (0, FRAME_SAMPLES - frame.size))
-        frame = np.ascontiguousarray(frame, dtype="<f4")
-        await client.send(
-            {
-                "type": "input_audio_buffer.append",
-                "audio": base64.b64encode(frame).decode("ascii"),
-                "format": "pcm_f32le",
-                "sample_rate_hz": SAMPLE_RATE_HZ,
-                "duration_ms": 80,
-                "audio_end_ms": (seq + 1) * 80,
-                "is_speech": bool(np.sqrt(np.mean(np.square(frame))) >= 1e-4),
-            }
-        )
+        await client.send(_frame_event(frame, seq))
         await asyncio.sleep(FRAME_PERIOD_S)
     return frame_count
 
 
+def _frame_event(frame: np.ndarray, seq: int) -> dict[str, object]:
+    frame = np.ascontiguousarray(np.pad(frame, (0, FRAME_SAMPLES - frame.size)), dtype="<f4")
+    return {
+        "type": "input_audio_buffer.append",
+        "audio": base64.b64encode(frame).decode("ascii"),
+        "format": "pcm_f32le",
+        "sample_rate_hz": SAMPLE_RATE_HZ,
+        "duration_ms": 80,
+        "audio_end_ms": (seq + 1) * 80,
+        "is_speech": bool(np.sqrt(np.mean(np.square(frame))) >= 1e-4),
+    }
+
+
 def _audio_frame_stats(
     raw: bytes,
     *,
@@ -242,11 +300,7 @@ def _audio_frame_stats(
 
 
 def _response_ids(client: RawRealtimeProbe) -> set[str]:
-    return {
-        str(event["response_id"])
-        for event in _events(client, "response.output_audio.delta")
-        if isinstance(event.get("response_id"), str)
-    }
+    return {str(event["response_id"]) for event in _audio_events(client) if isinstance(event.get("response_id"), str)}
 
 
 def _session_result(
@@ -256,40 +310,66 @@ def _session_result(
     args: argparse.Namespace,
     minimum_chunks: int,
 ) -> tuple[bytes, set[str], dict[str, object]]:
-    chunks = sum(len(items) for items in client.events.response_audio.values())
-    raw = client.events.audio_bytes()
-    if chunks < minimum_chunks or not raw or len(raw) % 2:
-        raise AssertionError(f"invalid PCM16 output: chunks={chunks}, bytes={len(raw)}")
+    if isinstance(client, RawRealtimeProbe):
+        audio_chunks = _validated_audio_chunks(client)
+        chunks = sum(bool(chunk) for _, chunk in audio_chunks)
+        raw = b"".join(chunk for _, chunk in audio_chunks if chunk)
+    else:
+        # Unit fixtures may provide an already-decoded EventCollector without
+        # retaining the base64 wire field. Real websocket probes always take
+        # the strict wire-validation path above.
+        chunks = sum(len(items) for items in client.events.response_audio.values())
+        raw = client.events.audio_bytes()
+    stats: dict[str, object] = {"audio_chunks": chunks, "audio_samples": len(raw) // 2}
+    if not raw or len(raw) % 2:
+        raise _AudioAcceptanceError("pcm16_output", f"invalid PCM16 output: chunks={chunks}, bytes={len(raw)}", stats)
     pcm = np.frombuffer(raw, dtype="<i2").astype(np.float32) / 32768.0
     rms = float(np.sqrt(np.mean(np.square(pcm))))
+    stats["audio_rms"] = rms
     # This low floor only catches corrupt, non-finite, or effectively zero PCM.
     # Audible speech is checked per frame below with the higher 1e-3 default.
     if not np.isfinite(pcm).all() or rms <= 1e-5:
-        raise AssertionError(f"output audio is non-finite or silent: samples={pcm.size}, rms={rms}")
-    rates = {
-        int(event["sample_rate_hz"])
-        for event in _events(client, "response.output_audio.delta")
-        if isinstance(event.get("sample_rate_hz"), int)
-    }
+        raise _AudioAcceptanceError(
+            "audio_rms", f"output audio is non-finite or silent: samples={pcm.size}, rms={rms}", stats
+        )
+    rates = {rate for event in _audio_events(client) if isinstance(rate := event.get("sample_rate_hz"), int)}
     if rates != {SAMPLE_RATE_HZ}:
-        raise AssertionError(f"unexpected output sample rates: {sorted(rates)}")
-    stats = _audio_frame_stats(
-        raw,
-        input_frames=input_frames,
-        voiced_frame_rms_threshold=args.voiced_frame_rms_threshold,
-    )
-    if not 0 <= int(stats["frame_deficit"]) <= args.max_frame_deficit:
-        raise AssertionError(f"invalid synchronous frame accounting: input={input_frames}, stats={stats}")
+        raise _AudioAcceptanceError("sample_rate", f"unexpected output sample rates: {sorted(rates)}", stats)
+    if input_frames < 1:
+        raise _AudioAcceptanceError("input_frames", f"audio received without input frames: input={input_frames}", stats)
+    try:
+        frame_stats = _audio_frame_stats(
+            raw,
+            input_frames=input_frames,
+            voiced_frame_rms_threshold=args.voiced_frame_rms_threshold,
+        )
+    except AssertionError as exc:
+        # _audio_frame_stats only formats local sample counts, not wire data.
+        raise _AudioAcceptanceError("whole_codec_frames", str(exc), stats) from exc
+    stats.update(frame_stats)
+    if chunks < minimum_chunks:
+        raise _AudioAcceptanceError(
+            "audio_chunks", f"insufficient audio chunks: chunks={chunks}, minimum={minimum_chunks}", stats
+        )
+    if not 0 <= int(frame_stats["frame_deficit"]) <= args.max_frame_deficit:
+        raise _AudioAcceptanceError(
+            "frame_deficit",
+            f"invalid synchronous frame accounting: input={input_frames}, maximum_deficit={args.max_frame_deficit}, "
+            f"stats={stats}",
+            stats,
+        )
     min_voiced_frames = int(args.min_voiced_frames)
     if min_voiced_frames < 1:
         raise ValueError(f"min_voiced_frames must be positive, got {min_voiced_frames}")
-    if int(stats["voiced_frames"]) < min_voiced_frames:
-        raise AssertionError(
-            f"output contained insufficient audible speech: minimum_voiced_frames={min_voiced_frames}, stats={stats}"
+    if int(frame_stats["voiced_frames"]) < min_voiced_frames:
+        raise _AudioAcceptanceError(
+            "voiced_frames",
+            f"output contained insufficient audible speech: minimum_voiced_frames={min_voiced_frames}, stats={stats}",
+            stats,
         )
     runtime_metadata = [
         metadata["vllm_omni"]
-        for event in _events(client, "response.output_audio.delta")
+        for event in _audio_events(client)
         if isinstance((metadata := event.get("metadata")), dict) and isinstance(metadata.get("vllm_omni"), dict)
     ]
     if not any(
@@ -298,18 +378,19 @@ def _session_result(
         and item.get("runner_kv_backed") is True
         for item in runtime_metadata
     ):
-        raise AssertionError(f"audio events did not prove the scheduler data plane: {runtime_metadata}")
+        raise _AudioAcceptanceError(
+            "scheduler_data_plane", "audio events did not prove the scheduler data plane", stats
+        )
     response_ids = _response_ids(client)
     if len(response_ids) != 1:
-        raise AssertionError(f"continuous stream used multiple responses: {sorted(response_ids)}")
+        raise _AudioAcceptanceError(
+            "response_identity", f"continuous stream requires one response: count={len(response_ids)}", stats
+        )
     return (
         raw,
         response_ids,
         {
             **stats,
-            "audio_chunks": chunks,
-            "audio_samples": len(raw) // 2,
-            "audio_rms": rms,
             "response_ids": sorted(response_ids),
         },
     )
@@ -332,6 +413,8 @@ def _save(output_dir: Path, name: str, client: RawRealtimeProbe, raw: bytes) ->
 
 
 async def run(args: argparse.Namespace) -> dict[str, object]:
+    if getattr(args, "sessions", None) is not None:
+        return await run_load(args)
     input_path = Path(args.input_wav)
     input_identity = _input_identity(
         input_path,
@@ -381,14 +464,12 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
     )
     await asyncio.gather(
         wait_for(
-            lambda: len(primary.events.audio_bytes()) // (2 * FRAME_SAMPLES) >= primary_frames - args.max_frame_deficit,
+            lambda: len(_audio_bytes(primary)) // (2 * FRAME_SAMPLES) >= primary_frames - args.max_frame_deficit,
             timeout_s=args.timeout_s,
             label="primary frame coverage",
         ),
         wait_for(
-            lambda: (
-                len(secondary.events.audio_bytes()) // (2 * FRAME_SAMPLES) >= secondary_frames - args.max_frame_deficit
-            ),
+            lambda: (len(_audio_bytes(secondary)) // (2 * FRAME_SAMPLES) >= secondary_frames - args.max_frame_deficit),
             timeout_s=args.timeout_s,
             label="secondary frame coverage",
         ),
@@ -426,16 +507,14 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
     await asyncio.gather(
         wait_for(
             lambda: (
-                len(secondary.events.audio_bytes()) // (2 * FRAME_SAMPLES)
-                >= secondary_total_frames - args.max_frame_deficit
+                len(_audio_bytes(secondary)) // (2 * FRAME_SAMPLES) >= secondary_total_frames - args.max_frame_deficit
             ),
             timeout_s=args.timeout_s,
             label="survivor frame coverage after slot recycle",
         ),
         wait_for(
             lambda: (
-                len(replacement.events.audio_bytes()) // (2 * FRAME_SAMPLES)
-                >= replacement_frames - args.max_frame_deficit
+                len(_audio_bytes(replacement)) // (2 * FRAME_SAMPLES) >= replacement_frames - args.max_frame_deficit
             ),
             timeout_s=args.timeout_s,
             label="replacement frame coverage",
@@ -493,6 +572,284 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
     return result
 
 
+def _distribution(values: list[float]) -> dict[str, int | float | None]:
+    if not values:
+        return {"count": 0, "median": None, "p99": None, "max": None, "argmax_index": None}
+    return {
+        "count": len(values),
+        "median": float(np.median(values)),
+        "p99": float(np.quantile(values, 0.99)),
+        "max": max(values),
+        "argmax_index": int(np.argmax(values)),
+    }
+
+
+def _load_metrics(client: RawRealtimeProbe, sends: list[tuple[float, float, float]]) -> dict[str, object]:
+    # Use the collector's local clock, never a timestamp supplied by the server.
+    packets: list[dict[str, object]] = []
+    response_ids: set[str] = set()
+    packet_times: list[float] = []
+    packet_bytes: list[int] = []
+    invalid_packets: list[dict[str, object]] = []
+    samples = 0
+    for index, (raw, received) in enumerate(zip(client.events.events, client.events.event_received_at_s, strict=True)):
+        if raw.get("type") not in AUDIO_DELTA_EVENT_TYPES:
+            continue
+        try:
+            chunk = _validated_audio_event(raw)
+        except ValueError as exc:
+            # Only fixed diagnostic codes from the validator are exported,
+            # never audio payloads or details from the remote endpoint.
+            invalid_packets.append({"event_index": index, "received_at_s": received, "code": str(exc)})
+            continue
+        if chunk:
+            samples += len(chunk) // 2
+            packet_times.append(received)
+            packet_bytes.append(len(chunk))
+            response_id = raw.get("response_id")
+            packets.append({"received_at_s": received, "samples": len(chunk) // 2, "response_id": response_id})
+            if isinstance(response_id, str) and response_id:
+                response_ids.add(response_id)
+    intervals = [(b - a) * 1000 for a, b in zip(packet_times, packet_times[1:])]
+    first_send = sends[0][1] if sends else None
+    audio_duration = samples / SAMPLE_RATE_HZ
+    first_audio_ms = (packet_times[0] - first_send) * 1000 if packet_times and first_send is not None else None
+    client_rtf = (packet_times[-1] - first_send) / audio_duration if audio_duration and first_send is not None else None
+    continuity = compute_continuity_stats(packet_times, packet_bytes, SAMPLE_RATE_HZ) if packet_times else None
+    lateness = _distribution([max(0.0, sent - planned) * 1000 for planned, sent, _ in sends])
+    p99_lateness = lateness["p99"]
+    return {
+        "input_frames": len(sends),
+        "output_samples": samples,
+        # Populate frame accounting from _session_result, including on failure.
+        "frame_deficit": None,
+        "response_ids": sorted(response_ids),
+        "client_first_audio_after_stream_start_ms": first_audio_ms,
+        "client_stream_rtf": client_rtf,
+        "client_audio_packet_interval_ms": _distribution(intervals),
+        "client_send_lateness_ms": lateness,
+        "client_pacing_warning": isinstance(p99_lateness, float) and p99_lateness > FRAME_PERIOD_S * 1000,
+        "audio_underrun_s": continuity.max_underrun_s if continuity else None,
+        "audio_underrun_event_count": continuity.underrun_event_count if continuity else None,
+        "audio_continuity_ok": continuity.is_continuous if continuity else None,
+        "input_timeline": [
+            {"frame_index": i, "planned_at_s": planned, "send_started_at_s": sent, "send_completed_at_s": completed}
+            for i, (planned, sent, completed) in enumerate(sends)
+        ],
+        "audio_packet_timeline": packets,
+        "invalid_audio_packets": invalid_packets,
+    }
+
+
+def _check_load_connection(client: RawRealtimeProbe) -> None:
+    if client.events.errors():
+        raise RuntimeError("server emitted a Realtime error (see the session error count)")
+    if client._reader_task is not None and client._reader_task.done():
+        client._reader_task.result()  # Propagate malformed JSON/reader failures.
+        raise RuntimeError("connection closed before session completion")
+
+
+async def _wait_load_event(client: RawRealtimeProbe, event_type: str, timeout_s: float) -> None:
+    def received() -> bool:
+        if client.events.count(event_type):
+            return True
+        _check_load_connection(client)
+        return False
+
+    await wait_for(received, timeout_s=timeout_s, label=event_type)
+
+
+async def _paced_load_frames(
+    client: RawRealtimeProbe,
+    pcm: np.ndarray,
+    *,
+    epoch: float,
+    timeout_s: float,
+    sends: list[tuple[float, float, float]],
+    clock: Callable[[], float] = time.monotonic,
+    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
+) -> None:
+    next_send = epoch
+    for seq in range(math.ceil(pcm.size / FRAME_SAMPLES)):
+        await sleep(max(0.0, next_send - clock()))
+        _check_load_connection(client)
+        event = _frame_event(pcm[seq * FRAME_SAMPLES : (seq + 1) * FRAME_SAMPLES], seq)
+        started = clock()
+        await asyncio.wait_for(client.send(event), timeout=timeout_s)
+        completed = clock()
+        sends.append((epoch + seq * FRAME_PERIOD_S, started, completed))
+        # Do not burst to catch up after a stalled send; report accumulated lag.
+        next_send = max(epoch + (seq + 1) * FRAME_PERIOD_S, completed + FRAME_PERIOD_S)
+
+
+async def _request_load_close(client: RawRealtimeProbe, timeout_s: float) -> None:
+    await client.send({"type": "session.close"})
+    await _wait_load_event(client, "session.closed", timeout_s)
+
+
+async def _run_load_session(
+    args: argparse.Namespace,
+    pcm: np.ndarray,
+    index: int,
+    ready: asyncio.Future[None],
+    start: asyncio.Future[float],
+) -> dict[str, object]:
+    session_id = f"personaplex-load-{index}-{uuid.uuid4().hex}"
+    client = RawRealtimeProbe(_realtime_url(args.url, args.model, session_id), close_timeout_s=args.cleanup_timeout_s)
+    sends: list[tuple[float, float, float]] = []
+    error: str | None = None
+    cleanup_error: str | None = None
+    metrics: dict[str, object] | None = None
+    session_stats: dict[str, object] = {}
+    acceptance_check: str | None = None
+    phase = "admission"
+    try:
+        async with client:
+            try:
+                await asyncio.wait_for(
+                    client.send(_session_update(args, session_id=session_id, persona=args.persona)), args.timeout_s
+                )
+                await _wait_load_event(client, "session.created", args.timeout_s)
+                capabilities = _capabilities(_events(client, "session.created")[-1])
+                if (
+                    capabilities.get("chunk_period_ms") != 80
+                    or capabilities.get("supports_multi_session_same_replica") is not True
+                ):
+                    raise AssertionError("server did not advertise PersonaPlex multi-session capabilities")
+                ready.set_result(None)
+                phase = "stream"
+                epoch = await asyncio.shield(start)
+                await _paced_load_frames(client, pcm, epoch=epoch, timeout_s=args.timeout_s, sends=sends)
+                phase = "drain"
+                # Fixed, reported observation window. Closing must not extend
+                # it and let a slow session pass by flushing after the cutoff.
+                await asyncio.sleep(args.drain_s)
+                _check_load_connection(client)
+                _, _, session_stats = _session_result(
+                    client, input_frames=len(sends), args=args, minimum_chunks=args.minimum_audio_chunks
+                )
+            finally:
+                metrics = _load_metrics(client, sends)
+                if client.events.count("session.created") and not client.events.count("session.closed"):
+                    try:
+                        # Send and acknowledgement share one short cleanup budget,
+                        # independent of the potentially long inference timeout.
+                        await asyncio.wait_for(
+                            _request_load_close(client, args.cleanup_timeout_s), args.cleanup_timeout_s
+                        )
+                    except (OSError, asyncio.TimeoutError, ValueError, RuntimeError, WebSocketException) as exc:
+                        cleanup_error = f"cleanup: {type(exc).__name__}"
+    except (OSError, asyncio.TimeoutError, ValueError, RuntimeError, AssertionError, WebSocketException) as exc:
+        if isinstance(exc, _AudioAcceptanceError):
+            acceptance_check = exc.check
+            session_stats = exc.stats
+            error = f"{phase}: AssertionError: {exc}"
+        else:
+            # Other assertions/errors may embed handshake credentials or wire
+            # metadata. Only our local numeric acceptance diagnostics are safe.
+            error = f"{phase}: {type(exc).__name__}"
+    finally:
+        # Admission failure must not leave the other sessions at the barrier.
+        if not ready.done():
+            ready.set_result(None)
+    if metrics is None:
+        metrics = _load_metrics(client, sends)
+    if cleanup_error:
+        error = f"{error}; {cleanup_error}" if error else cleanup_error
+    if client.events.errors():
+        error = error or "server emitted a Realtime error"
+    if metrics["invalid_audio_packets"]:
+        error = error or "invalid audio packets"
+    first_audio = metrics["client_first_audio_after_stream_start_ms"]
+    if isinstance(first_audio, float) and first_audio < 0:
+        error = error or "audio arrived before the first input frame"
+    rtf = metrics["client_stream_rtf"]
+    if args.max_client_rtf is not None and isinstance(rtf, float) and rtf > args.max_client_rtf:
+        error = error or f"client_stream_rtf exceeds {args.max_client_rtf}"
+    return {
+        "index": index,
+        "session_id": session_id,
+        "ok": error is None,
+        "error": error,
+        "acceptance_check": acceptance_check,
+        "server_error_count": len(client.events.errors()),
+        "server_error_codes": [
+            str(body.get("code", "unknown")) if isinstance(body := event.get("error"), dict) else "unknown"
+            for event in client.events.errors()
+        ],
+        **metrics,
+        **session_stats,
+    }
+
+
+async def run_load(args: argparse.Namespace) -> dict[str, object]:
+    input_path = Path(args.input_wav)
+    identity = _input_identity(input_path, expected_sha256=args.expected_input_sha256)
+    pcm = _read_wav_as_float32(input_path)
+    if args.tail_s:
+        pcm = np.concatenate([pcm, np.zeros(round(args.tail_s * SAMPLE_RATE_HZ), dtype=np.float32)])
+    pcm = pcm[: args.load_frames * FRAME_SAMPLES]
+    output_dir = Path(args.output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    # Refuse to silently overwrite evidence from a previous sweep.
+    result_path = output_dir / "load-result.json"
+    if result_path.exists():
+        raise FileExistsError(result_path)
+    loop = asyncio.get_running_loop()
+    start: asyncio.Future[float] = loop.create_future()
+    ready: list[asyncio.Future[None]] = [loop.create_future() for _ in range(args.sessions)]
+    tasks = [asyncio.create_task(_run_load_session(args, pcm, i, r, start)) for i, r in enumerate(ready)]
+    try:
+        await asyncio.gather(*ready)
+        start.set_result(time.monotonic())
+        rows = await asyncio.gather(*tasks)
+    finally:
+        for task in tasks:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+    seen: set[str] = set()
+    shared: set[str] = set()
+    for row in rows:
+        assert isinstance(row["response_ids"], list)
+        response_ids = set(row["response_ids"])
+        shared.update(seen & response_ids)
+        seen.update(response_ids)
+    for row in rows:
+        assert isinstance(row["response_ids"], list)
+        if shared.intersection(row["response_ids"]):
+            row.update(ok=False, error="concurrent sessions shared a response id")
+    result = {
+        "schema_version": 1,
+        "mode": "paced_multi_session",
+        "model": args.model,
+        "requested_sessions": args.sessions,
+        "passed_sessions": sum(row["ok"] is True for row in rows),
+        "ok": all(row["ok"] is True for row in rows),
+        "input": {**identity, "tail_s": args.tail_s, "load_frames": args.load_frames},
+        "frame_period_s": FRAME_PERIOD_S,
+        "drain_s": args.drain_s,
+        "cleanup_timeout_s": args.cleanup_timeout_s,
+        "client_pacing_warning_threshold_ms": FRAME_PERIOD_S * 1000,
+        "client_pacing_warning_sessions": [row["index"] for row in rows if row["client_pacing_warning"]],
+        "max_client_rtf": args.max_client_rtf,
+        "acceptance": {
+            "max_frame_deficit": args.max_frame_deficit,
+            "min_voiced_frames": args.min_voiced_frames,
+            "minimum_audio_chunks": args.minimum_audio_chunks,
+            "voiced_frame_rms_threshold": args.voiced_frame_rms_threshold,
+        },
+        "shared_response_ids": sorted(shared),
+        "sessions": rows,
+        "reported_server": {"revision": args.server_revision, "hardware": args.server_hardware},
+        "measurement_origin": "client monotonic clock; admission excluded; streaming and drain retained",
+        "interval_definition": "nonempty audio packet arrivals, not server ticks or per-frame inference latency",
+    }
+    with result_path.open("x", encoding="utf-8") as stream:
+        json.dump(result, stream, ensure_ascii=False, indent=2, allow_nan=False)
+    return result
+
+
 def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser()
     parser.add_argument("--url", default="ws://127.0.0.1:8099/v1/realtime?duplex=1")
@@ -510,6 +867,12 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--tail-s", type=float, default=2.0)
     parser.add_argument("--drain-s", type=float, default=2.0)
     parser.add_argument("--timeout-s", type=float, default=180.0)
+    parser.add_argument(
+        "--cleanup-timeout-s",
+        type=float,
+        default=5.0,
+        help="Load mode: shared session-close/ack budget; also bounds the websocket close handshake.",
+    )
     parser.add_argument(
         "--max-frame-deficit",
         type=int,
@@ -536,11 +899,39 @@ def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
         default=0,
         help="Frames sent after slot reuse; 0 replays the full input so the replacement must produce audible output.",
     )
-    return parser.parse_args(argv)
+    parser.add_argument(
+        "--sessions", type=int, help="Opt into paced load mode with N sessions; omit for the lifecycle test."
+    )
+    parser.add_argument(
+        "--load-frames", type=int, default=1000, help="Maximum input frames per load session (no repetition)."
+    )
+    parser.add_argument("--max-client-rtf", type=float, help="Optional client stream-window RTF acceptance ceiling.")
+    parser.add_argument("--server-revision", help="User-reported server commit/version; not verified by this client.")
+    parser.add_argument(
+        "--server-hardware", help="User-reported server hardware; not inferred from the client machine."
+    )
+    args = parser.parse_args(argv)
+    if args.sessions is not None:
+        if not 1 <= args.sessions <= 64 or not 1 <= args.load_frames <= 10000:
+            parser.error("load mode requires sessions in [1,64] and load-frames in [1,10000]")
+        for name in ("timeout_s", "cleanup_timeout_s", "drain_s", "tail_s"):
+            value = getattr(args, name)
+            if not math.isfinite(value) or value < 0 or (name in ("timeout_s", "cleanup_timeout_s") and value == 0):
+                parser.error(f"{name} must be finite and non-negative (timeout must be positive)")
+        if args.max_client_rtf is not None and (not math.isfinite(args.max_client_rtf) or args.max_client_rtf <= 0):
+            parser.error("max-client-rtf must be finite and positive")
+        if args.max_frame_deficit < 0 or args.min_voiced_frames < 1 or args.minimum_audio_chunks < 1:
+            parser.error("frame-deficit must be non-negative; voiced frames and audio chunks must be positive")
+        if not math.isfinite(args.voiced_frame_rms_threshold) or args.voiced_frame_rms_threshold < 0:
+            parser.error("voiced-frame-rms-threshold must be finite and non-negative")
+    return args
 
 
 def main() -> None:
-    print(json.dumps(asyncio.run(run(parse_args())), ensure_ascii=False, indent=2))
+    result = asyncio.run(run(parse_args()))
+    print(json.dumps(result, ensure_ascii=False, indent=2, allow_nan=False))
+    if result.get("mode") == "paced_multi_session" and not result["ok"]:
+        raise SystemExit(1)
 
 
 if __name__ == "__main__":

tests/e2e/online_serving/test_personaplex_load_driver.py

@@ -0,0 +1,568 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+import asyncio
+import base64
+import json
+import wave
+from contextlib import asynccontextmanager
+from typing import TypedDict
+
+import numpy as np
+import pytest
+import websockets
+
+from tests.e2e.online_serving import personaplex_realtime_duplex as driver
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class _ServerState(TypedDict):
+    index: int
+    closed: bool
+    frames: int
+    close_requested: bool
+
+
+def _args(tmp_path, *extra):
+    wav_path = tmp_path / "input.wav"
+    with wave.open(str(wav_path), "wb") as stream:
+        stream.setnchannels(1)
+        stream.setsampwidth(2)
+        stream.setframerate(driver.SAMPLE_RATE_HZ)
+        stream.writeframes(np.full(driver.FRAME_SAMPLES * 2, 2000, dtype="<i2").tobytes())
+    return driver.parse_args(
+        [
+            "--model",
+            "fixture-only",
+            "--input-wav",
+            str(wav_path),
+            "--output-dir",
+            str(tmp_path / "out"),
+            "--sessions",
+            "2",
+            "--load-frames",
+            "2",
+            "--tail-s",
+            "0",
+            "--drain-s",
+            "0.04",
+            "--timeout-s",
+            "3",
+            "--cleanup-timeout-s",
+            "1",
+            "--min-voiced-frames",
+            "1",
+            "--minimum-audio-chunks",
+            "1",
+            "--max-frame-deficit",
+            "0",
+            *extra,
+        ]
+    )
+
+
+def _audio(response_id="r", samples=1920, event_type="response.output_audio.delta"):
+    return {
+        "type": event_type,
+        "response_id": response_id,
+        "delta": base64.b64encode(np.full(samples, 2000, dtype="<i2").tobytes()).decode("ascii"),
+        "sample_rate_hz": driver.SAMPLE_RATE_HZ,
+        "metadata": {
+            "vllm_omni": {
+                "runtime_impl": "scheduler_data_plane",
+                "uses_model_runner_scheduler": True,
+                "runner_kv_backed": True,
+            }
+        },
+    }
+
+
+@asynccontextmanager
+async def _server(mode="success"):
+    states: list[_ServerState] = []
+
+    async def handle(ws):
+        index = len(states)
+        state: _ServerState = {"index": index, "closed": False, "frames": 0, "close_requested": False}
+        states.append(state)
+        response_id = "shared" if mode == "shared_response" else f"r-{index}"
+        try:
+            async for raw in ws:
+                message = json.loads(raw)
+                if message["type"] == "session.update":
+                    if mode == "reject_second" and index == 1:
+                        await ws.send(json.dumps({"type": "error", "error": {"code": "resource_exhausted"}}))
+                        continue
+                    if mode == "missing_capabilities":
+                        await ws.send(
+                            json.dumps(
+                                {
+                                    "type": "session.created",
+                                    "session": {"resume_token": "must-not-be-exported"},
+                                }
+                            )
+                        )
+                        continue
+                    await ws.send(
+                        json.dumps(
+                            {
+                                "type": "session.created",
+                                "session": {
+                                    "resume_token": "must-not-be-exported",
+                                    "capabilities": {
+                                        "chunk_period_ms": 80,
+                                        "supports_multi_session_same_replica": True,
+                                    },
+                                },
+                            }
+                        )
+                    )
+                    await ws.send(json.dumps({"type": "response.created", "response": {"id": response_id}}))
+                elif message["type"] == "input_audio_buffer.append":
+                    state["frames"] += 1
+                    if mode == "short_audio" and state["frames"] == 2:
+                        continue
+                    if mode == "disconnect":
+                        await ws.close()
+                        return
+                    if mode == "bad_json":
+                        await ws.send("{")
+                    elif mode not in ("silent", "late_on_close"):
+                        event_type = (
+                            "response.audio.delta" if mode == "legacy_audio_event" else "response.output_audio.delta"
+                        )
+                        packet = _audio(response_id, event_type=event_type)
+                        if mode == "private_metadata":
+                            packet["metadata"] = {"vllm_omni": {"private": "remote-detail-must-not-be-exported"}}
+                        if mode == "odd_packet":
+                            raw_audio = base64.b64decode(packet["delta"])
+                            raw_audio = raw_audio + b"\x00" if state["frames"] == 1 else raw_audio[:-1]
+                            packet["delta"] = base64.b64encode(raw_audio).decode("ascii")
+                        if state["frames"] == 2:
+                            if mode == "missing_packet_rate":
+                                packet.pop("sample_rate_hz")
+                            elif mode == "string_packet_rate":
+                                packet["sample_rate_hz"] = "24000"
+                            elif mode == "invalid_base64":
+                                packet["delta"] = "!" + packet["delta"]
+                            elif mode == "orphan_audio":
+                                orphan = _audio(response_id)
+                                orphan.pop("response_id")
+                                await ws.send(json.dumps(orphan))
+                        await ws.send(json.dumps(packet))
+                elif message["type"] == "session.close":
+                    state["close_requested"] = True
+                    if mode == "late_on_close":
+                        await ws.send(json.dumps(_audio(response_id, samples=driver.FRAME_SAMPLES * state["frames"])))
+                    if mode == "error_on_close":
+                        await ws.send(json.dumps({"type": "error", "error": {"code": "cleanup_failed"}}))
+                    if mode != "no_close_ack":
+                        await ws.send(json.dumps({"type": "session.closed"}))
+        finally:
+            state["closed"] = True
+
+    async with websockets.serve(handle, "127.0.0.1", 0) as server:
+        port = server.sockets[0].getsockname()[1]
+        yield f"ws://127.0.0.1:{port}/v1/realtime", states
+
+
+def test_omitted_sessions_preserves_lifecycle_mode():
+    args = driver.parse_args(["--model", "unused", "--input-wav", "unused.wav"])
+    assert args.sessions is None
+
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        ["--sessions", "0"],
+        ["--sessions", "65"],
+        ["--load-frames", "0"],
+        ["--load-frames", "10001"],
+        ["--drain-s", "nan"],
+        ["--tail-s", "-1"],
+        ["--timeout-s", "0"],
+        ["--timeout-s", "inf"],
+        ["--cleanup-timeout-s", "0"],
+        ["--cleanup-timeout-s", "nan"],
+        ["--max-client-rtf", "-1"],
+        ["--max-frame-deficit", "-1"],
+        ["--minimum-audio-chunks", "0"],
+        ["--min-voiced-frames", "0"],
+        ["--voiced-frame-rms-threshold", "nan"],
+    ],
+)
+def test_load_rejects_invalid_configuration(tmp_path, options):
+    with pytest.raises(SystemExit):
+        _args(tmp_path, *options)
+
+
+def test_metric_clock_and_packet_frame_distinction():
+    client = driver.RawRealtimeProbe("ws://unused")
+    client.events.add({"type": "session.created", "session": {"resume_token": "secret"}}, received_at_s=7)
+    for when in (10.4, 10.8, 11.6):
+        event = _audio(samples=driver.FRAME_SAMPLES * 5)
+        event["_client_received_at_s"] = -999.0
+        client.events.add(event, received_at_s=when)
+    report = driver._load_metrics(client, [(10, 10, 10.01)])
+    intervals = report["client_audio_packet_interval_ms"]
+    assert isinstance(intervals, dict)
+    assert intervals["count"] == 2
+    assert intervals["median"] == pytest.approx(600)
+    assert intervals["p99"] == pytest.approx(796)
+    assert intervals["max"] == pytest.approx(800)
+    assert intervals["argmax_index"] == 1
+    assert report["client_first_audio_after_stream_start_ms"] == pytest.approx(400)
+    assert report["output_samples"] == 15 * driver.FRAME_SAMPLES
+    assert report["audio_underrun_s"] == pytest.approx(0.4)
+    assert report["audio_continuity_ok"] is False
+    underruns = report["audio_underrun_event_count"]
+    assert isinstance(underruns, int) and underruns >= 1
+    assert report["client_pacing_warning"] is False
+    assert "secret" not in json.dumps(report)
+    assert "-999" not in json.dumps(report)
+
+
+def test_empty_output_has_missing_not_zero_latency():
+    report = driver._load_metrics(driver.RawRealtimeProbe("ws://unused"), [])
+    assert report["client_stream_rtf"] is None
+    assert report["audio_continuity_ok"] is None
+    assert report["audio_underrun_s"] is None
+    intervals = report["client_audio_packet_interval_ms"]
+    assert isinstance(intervals, dict)
+    assert intervals["p99"] is None
+
+
+def test_failed_acceptance_retains_frame_counts_without_remote_data():
+    client = driver.RawRealtimeProbe("ws://unused")
+    client.events.add(_audio(), received_at_s=1.0)
+    args = driver.parse_args(
+        ["--model", "unused", "--input-wav", "unused.wav", "--max-frame-deficit", "0", "--min-voiced-frames", "1"]
+    )
+    with pytest.raises(AssertionError) as raised:
+        driver._session_result(client, input_frames=2, args=args, minimum_chunks=1)
+    assert getattr(raised.value, "check", None) == "frame_deficit"
+    stats = getattr(raised.value, "stats", {})
+    assert stats["frame_deficit"] == 1 and type(stats["frame_deficit"]) is int
+    assert stats["output_frames"] == stats["voiced_frames"] == 1
+    assert stats["frame_coverage_ratio"] == 0.5
+
+
+@pytest.mark.parametrize("event_type", sorted(driver.AUDIO_DELTA_EVENT_TYPES))
+def test_session_result_accepts_current_and_legacy_audio_event_names(tmp_path, event_type):
+    client = driver.RawRealtimeProbe("ws://unused")
+    client.events.add(_audio(samples=driver.FRAME_SAMPLES, event_type=event_type), received_at_s=1.0)
+    raw, response_ids, stats = driver._session_result(
+        client,
+        input_frames=1,
+        args=_args(tmp_path, "--sessions", "1"),
+        minimum_chunks=1,
+    )
+    assert len(raw) == driver.FRAME_SAMPLES * 2
+    assert response_ids == {"r"}
+    assert stats["frame_deficit"] == 0
+
+
+def test_legacy_audio_event_name_contributes_load_metrics():
+    client = driver.RawRealtimeProbe("ws://unused")
+    client.events.add(_audio(samples=driver.FRAME_SAMPLES * 5, event_type="response.audio.delta"), received_at_s=10.4)
+    report = driver._load_metrics(client, [(10.0, 10.0, 10.01)])
+    assert report["output_samples"] == driver.FRAME_SAMPLES * 5
+    assert report["response_ids"] == ["r"]
+    assert report["invalid_audio_packets"] == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sessions", [1, 2, 4])
+async def test_real_websocket_load_success(tmp_path, sessions):
+    args = _args(tmp_path, "--sessions", str(sessions))
+    async with _server() as (url, states):
+        args.url = url
+        result = await driver.run(args)
+    assert result["ok"] is True
+    assert result["passed_sessions"] == sessions
+    rows = result["sessions"]
+    assert isinstance(rows, list)
+    assert len(rows) == sessions
+    assert all(row["frame_deficit"] == 0 for row in rows)
+    for row in rows:
+        assert type(row["frame_deficit"]) is int
+        assert row["output_frames"] == row["voiced_frames"] == row["audio_chunks"] == 2
+        assert row["silent_frames"] == 0
+        assert row["frame_coverage_ratio"] == 1.0
+        assert row["audio_rms"] == pytest.approx(2000 / 32768)
+        assert row["acceptance_check"] is None
+        assert isinstance(row["audio_continuity_ok"], bool)
+    assert all(s["closed"] and s["close_requested"] for s in states)
+    assert "must-not-be-exported" not in json.dumps(result)
+    saved = json.loads((tmp_path / "out/load-result.json").read_text())
+    assert saved == result
+    origins = [row["input_timeline"][0]["planned_at_s"] for row in rows]
+    assert len(set(origins)) == 1
+
+
+@pytest.mark.asyncio
+async def test_real_websocket_load_accepts_legacy_audio_event_name(tmp_path):
+    args = _args(tmp_path, "--sessions", "2")
+    async with _server("legacy_audio_event") as (url, states):
+        args.url = url
+        result = await driver.run(args)
+    assert result["ok"] is True
+    assert result["passed_sessions"] == 2
+    assert result["shared_response_ids"] == []
+    assert all(s["closed"] and s["close_requested"] for s in states)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "mode",
+    [
+        "reject_second",
+        "silent",
+        "disconnect",
+        "no_close_ack",
+        "shared_response",
+        "late_on_close",
+        "error_on_close",
+        "bad_json",
+        "missing_capabilities",
+    ],
+)
+async def test_real_websocket_failures_remain_in_results(tmp_path, mode):
+    args = _args(tmp_path)
+    async with _server(mode) as (url, states):
+        args.url = url
+        result = await asyncio.wait_for(driver.run(args), timeout=4)
+    assert result["ok"] is False
+    assert "must-not-be-exported" not in json.dumps(result)
+    assert result["requested_sessions"] == 2
+    rows = result["sessions"]
+    assert isinstance(rows, list)
+    assert len(rows) == 2
+    assert all(s["closed"] for s in states)
+    if mode == "reject_second":
+        assert result["passed_sessions"] == 1
+        rejected = next(row for row in rows if not row["ok"])
+        assert rejected["input_frames"] == 0
+        assert rejected["client_stream_rtf"] is None
+        assert rejected["server_error_codes"] == ["resource_exhausted"]
+    if mode == "silent":
+        # A failed measurement must still release its admitted session.
+        assert all(s["close_requested"] for s in states)
+    if mode == "late_on_close":
+        assert all(row["output_samples"] == 0 for row in rows)
+    if mode == "shared_response":
+        assert result["passed_sessions"] == 0
+
+
+@pytest.mark.asyncio
+async def test_existing_report_is_not_overwritten(tmp_path):
+    args = _args(tmp_path)
+    path = tmp_path / "out/load-result.json"
+    path.parent.mkdir()
+    path.write_text("previous measurement")
+    with pytest.raises(FileExistsError):
+        await driver.run(args)
+    assert path.read_text() == "previous measurement"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["success", "no_close_ack"])
+async def test_cancellation_closes_all_connections(tmp_path, mode):
+    args = _args(tmp_path, "--drain-s", "10", "--timeout-s", "30")
+    async with _server(mode) as (url, states):
+        args.url = url
+        task = asyncio.create_task(driver.run(args))
+
+        async def started():
+            while len(states) < 2 or not all(state["frames"] for state in states):
+                await asyncio.sleep(0.01)
+
+        await asyncio.wait_for(started(), timeout=3)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=3)
+    assert all(s["closed"] for s in states)
+
+
+@pytest.mark.asyncio
+async def test_client_rtf_ceiling_marks_slow_sessions_failed(tmp_path):
+    args = _args(tmp_path, "--max-client-rtf", "0.001")
+    async with _server() as (url, _):
+        args.url = url
+        result = await driver.run(args)
+    assert result["passed_sessions"] == 0
+    rows = result["sessions"]
+    assert isinstance(rows, list)
+    assert all("client_stream_rtf exceeds" in row["error"] for row in rows)
+
+
+@pytest.mark.asyncio
+async def test_stalled_send_does_not_cause_catchup_burst():
+    class Clock:
+        now = 10.0
+
+        def monotonic(self):
+            return self.now
+
+        async def sleep(self, delay):
+            self.now += delay
+
+    class Client(driver.RawRealtimeProbe):
+        async def send(self, event):
+            if event["audio_end_ms"] == 80:
+                clock.now += 0.25
+
+    clock = Clock()
+    sends: list[tuple[float, float, float]] = []
+    await driver._paced_load_frames(
+        Client("ws://unused"),
+        np.zeros(driver.FRAME_SAMPLES * 3, dtype="<f4"),
+        epoch=10.0,
+        timeout_s=1.0,
+        sends=sends,
+        clock=clock.monotonic,
+        sleep=clock.sleep,
+    )
+    assert len(sends) == 3
+    assert sends[1][1] >= sends[0][2] + driver.FRAME_PERIOD_S
+    assert sends[1][1] - sends[1][0] == pytest.approx(0.25)
+    assert sends[2][1] >= sends[1][2] + driver.FRAME_PERIOD_S
+    metrics = driver._load_metrics(Client("ws://unused"), sends)
+    assert metrics["client_pacing_warning"] is True
+    assert asyncio.sleep != clock.sleep
+
+
+@pytest.mark.asyncio
+async def test_failed_frame_acceptance_preserves_numeric_diagnostics(tmp_path):
+    args = _args(tmp_path, "--sessions", "1")
+    async with _server("short_audio") as (url, states):
+        args.url = url
+        result = await driver.run(args)
+    rows = result["sessions"]
+    assert isinstance(rows, list)
+    row = rows[0]
+    assert row["ok"] is False
+    assert row["acceptance_check"] == "frame_deficit"
+    assert row["frame_deficit"] == 1 and type(row["frame_deficit"]) is int
+    assert row["output_frames"] == row["voiced_frames"] == row["audio_chunks"] == 1
+    assert row["frame_coverage_ratio"] == 0.5
+    assert row["audio_rms"] == pytest.approx(2000 / 32768)
+    assert "input=2" in row["error"] and "maximum_deficit=0" in row["error"]
+    assert all(state["closed"] and state["close_requested"] for state in states)
+    assert json.loads((tmp_path / "out/load-result.json").read_text()) == result
+
+
+@pytest.mark.asyncio
+async def test_acceptance_failure_does_not_export_remote_metadata(tmp_path):
+    args = _args(tmp_path, "--sessions", "1")
+    async with _server("private_metadata") as (url, _):
+        args.url = url
+        result = await driver.run(args)
+    rows = result["sessions"]
+    assert isinstance(rows, list)
+    row = rows[0]
+    assert row["ok"] is False
+    assert row["acceptance_check"] == "scheduler_data_plane"
+    assert "remote-detail" not in json.dumps(result)
+    assert "must-not-be-exported" not in json.dumps(result)
+
+
+@pytest.mark.parametrize("ok", [True, False])
+def test_load_main_exit_status(tmp_path, monkeypatch, ok):
+    args = _args(tmp_path)
+
+    async def completed(_args):
+        return {"mode": "paced_multi_session", "ok": ok}
+
+    monkeypatch.setattr(driver, "parse_args", lambda: args)
+    monkeypatch.setattr(driver, "run", completed)
+    if ok:
+        driver.main()
+    else:
+        with pytest.raises(SystemExit) as exc:
+            driver.main()
+        assert exc.value.code == 1
+
+
+@pytest.mark.asyncio
+async def test_handshake_rejection_is_retained_without_server_details(tmp_path, monkeypatch):
+    from websockets.exceptions import InvalidHandshake
+
+    async def rejected(_self):
+        raise InvalidHandshake("credentials-in-remote-handshake-detail")
+
+    monkeypatch.setattr(driver.RawRealtimeProbe, "__aenter__", rejected)
+    result = await asyncio.wait_for(driver.run(_args(tmp_path)), timeout=3)
+    assert result["ok"] is False
+    rows = result["sessions"]
+    assert isinstance(rows, list) and len(rows) == 2
+    assert all(row["error"] == "admission: InvalidHandshake" for row in rows)
+    assert "credentials-in-remote-handshake-detail" not in json.dumps(result)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sessions", [1, 2])
+@pytest.mark.parametrize(
+    "mode",
+    [
+        "missing_packet_rate",
+        "string_packet_rate",
+        "invalid_base64",
+        "orphan_audio",
+        "odd_packet",
+    ],
+)
+async def test_load_rejects_invalid_individual_audio_packets(tmp_path, sessions, mode):
+    # The concatenated PCM can still meet coverage/voicing thresholds, so the
+    # validator must also check each packet before accepting the whole run.
+    args = _args(tmp_path, "--sessions", str(sessions))
+    async with _server(mode) as (url, states):
+        args.url = url
+        result = await asyncio.wait_for(driver.run(args), timeout=4)
+    assert result["ok"] is False
+    assert result["passed_sessions"] == 0
+    rows = result["sessions"]
+    assert isinstance(rows, list) and len(rows) == sessions
+    assert all(row["error"] is not None for row in rows)
+    expected_code = {
+        "missing_packet_rate": "invalid_audio_sample_rate",
+        "string_packet_rate": "invalid_audio_sample_rate",
+        "invalid_base64": "invalid_audio_base64",
+        "orphan_audio": "missing_audio_response_id",
+        "odd_packet": "unaligned_pcm16_packet",
+    }[mode]
+    assert all({packet["code"] for packet in row["invalid_audio_packets"]} == {expected_code} for row in rows)
+    assert all(
+        sum(packet["samples"] for packet in row["audio_packet_timeline"]) == row["output_samples"] for row in rows
+    )
+    assert all(state["closed"] and state["close_requested"] for state in states)
+    assert json.loads((tmp_path / "out/load-result.json").read_text()) == result
+
+
+def test_load_allows_empty_and_multi_frame_audio_packets():
+    client = driver.RawRealtimeProbe("ws://unused")
+    client.events.add(_audio(samples=0), received_at_s=1.0)
+    client.events.add(_audio(samples=driver.FRAME_SAMPLES * 5), received_at_s=1.4)
+    metrics = driver._load_metrics(client, [(1.0, 1.0, 1.001)])
+    assert metrics["invalid_audio_packets"] == []
+    assert metrics["output_samples"] == driver.FRAME_SAMPLES * 5
+    packets = metrics["audio_packet_timeline"]
+    assert isinstance(packets, list) and len(packets) == 1
+
+
+def test_load_packet_diagnostics_do_not_copy_remote_payload():
+    client = driver.RawRealtimeProbe("ws://unused")
+    event = _audio()
+    event["delta"] = "private-remote-payload!"
+    client.events.add(event, received_at_s=1.4)
+    metrics = driver._load_metrics(client, [])
+    assert metrics["output_samples"] == 0
+    assert metrics["invalid_audio_packets"] == [
+        {
+            "event_index": 0,
+            "received_at_s": 1.4,
+            "code": "invalid_audio_base64",
+        }
+    ]
+    assert "private-remote-payload" not in json.dumps(metrics)

vllm_omni/diffusion/layers/indexed_modulation.py

@@ -24,6 +24,18 @@ def _launch_row_chunks(kernel, rows: int, device_type: str, *args, **kwargs) ->
         kernel[(chunk_rows,)](*args, row_offset, **kwargs)
 
 
+def _use_hopper_bf16_affine_semantics(x: torch.Tensor) -> bool:
+    """Use the unfused AdaLN dtype boundaries on Hopper.
+
+    The fused FP32 modulation path is stable on Blackwell, but the SM90
+    kernel can accumulate enough extra precision in the repeated RMSNorm and
+    AdaLN operations to move the 50-step I2VA result below its CI threshold.
+    Keep the existing path on other architectures until their numerical
+    baselines are independently validated.
+    """
+    return x.device.type == "cuda" and torch.cuda.get_device_capability(x.device)[0] == 9
+
+
 @triton.jit
 def _indexed_scale_shift_kernel(
     output_ptr,
@@ -102,6 +114,7 @@ def _rms_norm_indexed_scale_shift_kernel(
     stride_scale_row,
     stride_indices,
     row_offset,
+    use_bf16_affine: tl.constexpr,
     block_n: tl.constexpr,
 ):
     row = tl.program_id(0) + row_offset
@@ -112,13 +125,20 @@ def _rms_norm_indexed_scale_shift_kernel(
     x = tl.load(x_ptr + row * stride_x_row + columns, mask=mask, other=0.0).to(tl.float32)
     weight = tl.load(weight_ptr + columns, mask=mask, other=0.0).to(tl.float32)
     variance = tl.sum(x * x, axis=0) / hidden_size
-    normalized = x * tl.rsqrt(variance + eps) * weight
-
-    shift = tl.load(shift_ptr + index * stride_shift_row + columns, mask=mask, other=0.0).to(tl.float32)
-    scale = tl.load(scale_ptr + index * stride_scale_row + columns, mask=mask, other=0.0).to(tl.float32)
+    if use_bf16_affine:
+        # Match the unfused RMSNorm output consumed by AdaLN on Hopper.
+        normalized = (x * tl.rsqrt(variance + eps) * weight).to(tl.bfloat16)
+        shift = tl.load(shift_ptr + index * stride_shift_row + columns, mask=mask, other=0.0)
+        scale = tl.load(scale_ptr + index * stride_scale_row + columns, mask=mask, other=0.0)
+        output = normalized * (scale + 1.0) + shift
+    else:
+        normalized = x * tl.rsqrt(variance + eps) * weight
+        shift = tl.load(shift_ptr + index * stride_shift_row + columns, mask=mask, other=0.0).to(tl.float32)
+        scale = tl.load(scale_ptr + index * stride_scale_row + columns, mask=mask, other=0.0).to(tl.float32)
+        output = normalized * (1.0 + scale) + shift
     tl.store(
         output_ptr + row * hidden_size + columns,
-        normalized * (1.0 + scale) + shift,
+        output,
         mask=mask,
     )
 
@@ -143,6 +163,7 @@ def _indexed_gate_rms_norm_scale_shift_kernel(
     stride_scale_row,
     stride_indices,
     row_offset,
+    use_bf16_affine: tl.constexpr,
     block_n: tl.constexpr,
 ):
     row = tl.program_id(0) + row_offset
@@ -159,12 +180,20 @@ def _indexed_gate_rms_norm_scale_shift_kernel(
 
     weight = tl.load(weight_ptr + columns, mask=mask, other=0.0).to(tl.float32)
     variance = tl.sum(updated * updated, axis=0) / hidden_size
-    normalized = updated * tl.rsqrt(variance + eps) * weight
-    shift = tl.load(shift_ptr + index * stride_shift_row + columns, mask=mask, other=0.0).to(tl.float32)
-    scale = tl.load(scale_ptr + index * stride_scale_row + columns, mask=mask, other=0.0).to(tl.float32)
+    if use_bf16_affine:
+        # Match the unfused RMSNorm output consumed by AdaLN on Hopper.
+        normalized = (updated * tl.rsqrt(variance + eps) * weight).to(tl.bfloat16)
+        shift = tl.load(shift_ptr + index * stride_shift_row + columns, mask=mask, other=0.0)
+        scale = tl.load(scale_ptr + index * stride_scale_row + columns, mask=mask, other=0.0)
+        output = normalized * (scale + 1.0) + shift
+    else:
+        normalized = updated * tl.rsqrt(variance + eps) * weight
+        shift = tl.load(shift_ptr + index * stride_shift_row + columns, mask=mask, other=0.0).to(tl.float32)
+        scale = tl.load(scale_ptr + index * stride_scale_row + columns, mask=mask, other=0.0).to(tl.float32)
+        output = normalized * (1.0 + scale) + shift
     tl.store(
         modulated_out_ptr + row * hidden_size + columns,
-        normalized * (1.0 + scale) + shift,
+        output,
         mask=mask,
     )
 
@@ -271,6 +300,7 @@ def rms_norm_indexed_scale_shift(
             shift.stride(0),
             scale.stride(0),
             indices.stride(0),
+            use_bf16_affine=_use_hopper_bf16_affine_semantics(x),
             block_n=triton.next_power_of_2(hidden_size),
             num_warps=8,
         )
@@ -324,6 +354,7 @@ def indexed_gate_rms_norm_scale_shift(
             shift.stride(0),
             scale.stride(0),
             indices.stride(0),
+            use_bf16_affine=_use_hopper_bf16_affine_semantics(residual),
             block_n=triton.next_power_of_2(hidden_size),
             num_warps=8,
         )

vllm_omni/diffusion/models/minimax_h3/fasth3.py

@@ -102,6 +102,7 @@ _MODEL_LEVEL_TARGETS = {
     "time_embedder.linear_2": "time_embedder.proj_out",
     "norm_out.linear": "final_layer.adaln_proj.linear",
     "norm_out.norm": "final_layer.norm",
+    "token_refiner.final_norm": "token_refiner.final_norm",
 }
 
 # Per-block adapter suffix -> (native suffix, how a delta enters the native
@@ -115,6 +116,8 @@ _BLOCK_TARGETS = {
     "attn.to_v": ("attn.qkv_proj", _QKV_V),
     "attn.to_out.0": ("attn.out_proj", _PLAIN),
     "attn.to_gate_compress": ("attn.to_gate_compress", _PLAIN),
+    "attn.norm_q": ("attn.q_norm", _PLAIN),
+    "attn.norm_k": ("attn.k_norm", _PLAIN),
     "ff.net.0.proj": ("mlp.fc1", _SWAP_HALVES),
     "ff.net.2": ("mlp.fc2", _PLAIN),
     "adaln_proj.linear": ("adaln_proj.linear", _PLAIN),

vllm_omni/diffusion/models/minimax_h3/fasth3_checkpoint.py

@@ -0,0 +1,131 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""Sampling contract for FastVideo FastH3 full checkpoints."""
+
+from __future__ import annotations
+
+import json
+import math
+from collections.abc import Mapping
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+from vllm_omni.diffusion.sched.sigma_schedule import DMD2SigmaSchedule
+from vllm_omni.errors import OmniClientError
+from vllm_omni.model_executor.model_loader.weight_utils import download_weights_from_hf_specific
+
+from .fasth3 import FASTH3_BASE_MODEL, _resolve_dit_attention_backend
+
+if TYPE_CHECKING:
+    from vllm_omni.diffusion.data import OmniDiffusionConfig
+    from vllm_omni.inputs.data import OmniDiffusionSamplingParams
+
+FASTH3_V2_MODEL_ID = "FastVideo/FastVideo-FastH3-8-Step-V2"
+FASTH3_V2_BASE_SCHEDULE = DMD2SigmaSchedule.from_positions((0.999, 0.874, 0.749, 0.624, 0.5, 0.375, 0.25, 0.125, 0.0))
+
+
+@dataclass(frozen=True)
+class FastH3CheckpointSpec:
+    """The full V2 release has its own schedule and trained attention policy.
+
+    This is independent of ``FastH3WeightFusion``: the released weights already
+    contain the entire student, including its learned compression gates.
+    """
+
+    vsa_sparsity: float = 0.8
+
+    @classmethod
+    def from_metadata(cls, metadata: Mapping[str, object]) -> FastH3CheckpointSpec:
+        """Validate the release's own fastvideo_inference.json."""
+        required = {
+            "model_id": FASTH3_V2_MODEL_ID,
+            "schema_version": "fasth3-inference-contract-v1",
+            "dmd_denoising_steps": [999, 874, 749, 624, 500, 375, 250, 125],
+            "transformer_forwards": 8,
+            "video_scheduler_shift": 10.0,
+            "audio_scheduler_shift": 3.0,
+            "guidance_scale": 1.0,
+            "attention_backend": "VIDEO_SPARSE_ATTN_H3",
+            "vsa_sparsity": 0.8,
+            "vsa_tile_size": 64,
+            "task": "t2av",
+        }
+        for key, expected in required.items():
+            if metadata.get(key) != expected:
+                raise ValueError(f"unsupported FastH3 contract: {key}={metadata.get(key)!r}, expected {expected!r}")
+        return cls()
+
+    def release_metadata(self) -> dict[str, Any]:
+        """Express the sampling policy in the existing H3 pipeline schema."""
+        return {
+            "partition": "fl2va",
+            "tasks": ["t2va"],
+            "sigma_shift_scales": {"video": 10.0, "audio": 3.0},
+            "base_schedule": list(FASTH3_V2_BASE_SCHEDULE.base_schedule),
+        }
+
+    def resolve_native_vaes(self, model_root: Path) -> Path:
+        """Reuse the frozen base VAEs with Omni's native tiled/parallel runtime.
+
+        Only VAE components are fetched; the student and text encoder come
+        directly from the FastVideo release. The release pins the base revision.
+        """
+        provenance = json.loads((model_root / "provenance.json").read_text(encoding="utf-8"))
+        base = provenance.get("base_model", "")
+        prefix = f"hf://{FASTH3_BASE_MODEL}@"
+        if not isinstance(base, str) or not base.startswith(prefix) or not base[len(prefix) :]:
+            raise ValueError("FastH3 V2 provenance must pin a MiniMaxAI/MiniMax-H3 base revision")
+        return (
+            Path(
+                download_weights_from_hf_specific(
+                    model_name_or_path=FASTH3_BASE_MODEL,
+                    cache_dir=None,
+                    allow_patterns=["FL2VA/video_vae/**", "FL2VA/audio_vae/**"],
+                    revision=base[len(prefix) :],
+                    require_all=True,
+                )
+            )
+            / "FL2VA"
+        )
+
+    def check_serving_contract(self, *, partition: str, od_config: OmniDiffusionConfig) -> None:
+        if partition != "fl2va":
+            raise ValueError("FastH3 V2 requires --task-type fl2va (T2VA requests only)")
+        if getattr(od_config, "lora_path", None):
+            raise ValueError("FastH3 V2 is a full checkpoint; additional LoRA adapters are unsupported")
+        if _resolve_dit_attention_backend(od_config) != "FASTVIDEO_VSA":
+            raise ValueError("FastH3 V2 requires --diffusion-attention-backend FASTVIDEO_VSA")
+        attention_config = getattr(od_config, "diffusion_attention_config", None)
+        per_role = getattr(attention_config, "per_role", None) or {}
+        spec = per_role.get("self") or getattr(attention_config, "default", None)
+        if getattr(spec, "fastvideo_vsa_topk", None) is not None:
+            raise ValueError("FastH3 V2 pins VSA sparsity=0.8; remove the fixed fastvideo_vsa_topk override")
+        parallel = getattr(od_config, "parallel_config", None)
+        if any(int(getattr(parallel, key, 1) or 1) != 1 for key in ("ring_degree", "allgather_degree")):
+            raise ValueError("FastH3 V2 supports local attention or pure Ulysses sequence parallelism")
+
+    def check_request(self, sampling: OmniDiffusionSamplingParams, *, step_execution: bool = False) -> None:
+        if step_execution:
+            # StepScheduler admits requests before the pipeline hook runs and
+            # derives its lifetime from these fields.  The pinned schedule must
+            # therefore be explicit and cannot be overridden by custom arrays.
+            if sampling.num_inference_steps != FASTH3_V2_BASE_SCHEDULE.num_inference_steps:
+                raise OmniClientError("FastH3 V2 step execution requires num_inference_steps=8")
+            if sampling.timesteps is not None or sampling.sigmas is not None:
+                raise OmniClientError("FastH3 V2 step execution does not support custom timesteps or sigmas")
+        if sampling.lora_request is not None:
+            raise OmniClientError("FastH3 V2 does not support per-request LoRA adapters")
+        steps = sampling.num_inference_steps
+        if steps is not None and steps != FASTH3_V2_BASE_SCHEDULE.num_inference_steps:
+            raise OmniClientError("FastH3 V2 requires num_inference_steps=8 (nine sigma points), or omitted")
+        extra = sampling.extra_args or {}
+        for key, expected in (("flow_shift", 10.0), ("audio_flow_shift", 3.0)):
+            try:
+                value = float(extra.get(key, expected))
+            except (TypeError, ValueError) as exc:
+                raise OmniClientError(f"FastH3 V2 requires {key}={expected:g}") from exc
+            if not math.isclose(value, expected):
+                raise OmniClientError(f"FastH3 V2 requires {key}={expected:g}, got {value:g}")
+        if sampling.guidance_scale is not None and sampling.guidance_scale != 1.0:
+            raise OmniClientError("FastH3 V2 requires guidance_scale=1")

vllm_omni/diffusion/models/minimax_h3/minimax_h3_transformer.py

@@ -55,6 +55,8 @@ from vllm_omni.diffusion.layers.rope import RotaryEmbedding
 from vllm_omni.diffusion.models.host_weight_contract import FinalLayoutModelContract
 from vllm_omni.platforms import current_omni_platform
 
+from .fasth3 import _resolve_native_target
+
 if TYPE_CHECKING:
     from vllm.model_executor.layers.quantization.base_config import (
         QuantizationConfig,
@@ -118,6 +120,18 @@ class MiniMaxH3DiTArchConfig:
 
     @classmethod
     def from_mapping(cls, config: Mapping[str, Any]) -> MiniMaxH3DiTArchConfig:
+        # The modular Diffusers checkpoint uses these spellings for the same
+        # architecture. Normalize before constructing the existing native DiT.
+        aliases = {
+            "num_refiner_layers": "token_refiner_num_layers",
+            "ffn_dim": "ffn_hidden_size",
+            "in_channels": "latents_dim",
+            "audio_in_channels": "audio_latents_dim",
+            "freq_dim": "timestep_input_dim",
+            "time_embed_hidden_dim": "time_embed_hidden_size",
+            "rope_freq_dim": "rope_inv_freq_len",
+        }
+        config = {aliases.get(name, name): value for name, value in config.items()}
         fields = cls.__dataclass_fields__
         values = {name: config[name] for name in fields if name in config}
         if "patch_size" in values:
@@ -394,6 +408,9 @@ def _sdpa_varlen_attention(
 
 
 class MiniMaxH3Attention(nn.Module):
+    # Full sparse checkpoints pin a ratio; legacy adapters use backend top-k.
+    vsa_sparsity: float | None = None
+
     def __init__(
         self,
         arch: MiniMaxH3DiTArchConfig,
@@ -588,6 +605,7 @@ class MiniMaxH3Attention(nn.Module):
                 # (see MINIMAX_H3_LASER_INPUT_SCALE). Ignored by every other
                 # backend/path.
                 "laser_input_scale": MINIMAX_H3_LASER_INPUT_SCALE,
+                **({"vsa_h3_sparsity": self.vsa_sparsity} if self.vsa_sparsity is not None else {}),
                 # Present only for a VSA artifact; the VSA backend reads it as
                 # the learned compression gate and every other backend ignores it.
                 **({"gate_compress": gate_compress.unsqueeze(0)} if gate_compress is not None else {}),
@@ -1127,10 +1145,23 @@ class MiniMaxH3DiTModel(nn.Module):
         self,
         od_config: OmniDiffusionConfig,
         quant_config: QuantizationConfig | None = None,
+        *,
+        diffusers_weights: bool | None = None,
     ) -> None:
         super().__init__()
         tf_config = od_config.tf_model_config
         config_mapping = tf_config.to_dict() if hasattr(tf_config, "to_dict") else dict(tf_config)
+        # The native MiniMax-H3 Hub snapshot advertises the Diffusers
+        # transformer class in its root config, while its FL2VA/Ref2VA
+        # components still contain native weights.  The pipeline has already
+        # resolved the actual source format, so let it override the
+        # class-name heuristic.  Keep the heuristic for standalone callers.
+        self._diffusers_weights = (
+            config_mapping.get("_class_name") == "MiniMaxH3Transformer3DModel"
+            if diffusers_weights is None
+            else diffusers_weights
+        )
+        self._rope_theta = float(config_mapping.get("rope_theta", 10000.0))
         arch = MiniMaxH3DiTArchConfig.from_mapping(config_mapping)
         self.arch = arch
         self.od_config = od_config
@@ -1211,7 +1242,7 @@ class MiniMaxH3DiTModel(nn.Module):
         )
         self._mark_missing_params_required()
 
-    def enable_vsa_gates(self) -> None:
+    def enable_vsa_gates(self, *, sparsity: float | None = None) -> None:
         """Give every DiT block's attention a VSA compression gate.
 
         A FastH3 VSA artifact assigns these projections rather than adding to
@@ -1223,6 +1254,9 @@ class MiniMaxH3DiTModel(nn.Module):
             return
         for block in self.blocks:
             block.attn.enable_vsa_gate()
+            block.attn.vsa_sparsity = sparsity
+            if sparsity is not None:
+                block.attn.to_gate_compress.weight.missing_param_init = "error"
         self.vsa_gates_enabled = True
 
     def _mark_missing_params_required(self) -> None:
@@ -1293,17 +1327,37 @@ class MiniMaxH3DiTModel(nn.Module):
         self,
         weights: Iterable[tuple[str, torch.Tensor]],
     ) -> set[str]:
-        """Load exact H3 checkpoint names with logical TP-aware loaders."""
+        """Load native or Diffusers H3 weights with the existing TP-aware loaders."""
         params = dict(self.named_parameters())
         params.update(dict(self.named_buffers()))
         loaded: set[str] = set()
+        diffusers_weights = getattr(self, "_diffusers_weights", False)
+        qkv_parts: dict[str, set[str]] = {}
+        source_names: set[str] = set()
         for name, loaded_weight in weights:
+            layout = "plain"
+            if diffusers_weights:
+                if name in source_names:
+                    raise ValueError(f"duplicate Diffusers H3 weight: {name}")
+                source_names.add(name)
+                module, _, kind = name.rpartition(".")
+                target = _resolve_native_target(module)
+                if target is None or kind not in {"weight", "bias"}:
+                    raise ValueError(f"unsupported Diffusers H3 weight: {name}")
+                name, layout = f"{target[0]}.{kind}", target[1]
             param = params.get(name)
             if param is None:
+                if diffusers_weights:
+                    raise ValueError(f"Diffusers H3 weight has no model parameter: {name}")
                 logger.warning("Skipping MiniMax H3 weight not present in model: %s", name)
                 continue
             weight_loader = getattr(param, "weight_loader", default_weight_loader)
-            if name.endswith(".attn.qkv_proj.weight"):
+            if layout in {"q", "k", "v"}:
+                # vLLM can load each projection directly into its packed QKV
+                # parameter, including TP slicing and online quantization.
+                weight_loader(param, loaded_weight, layout)
+                qkv_parts.setdefault(name, set()).add(layout)
+            elif name.endswith(".attn.qkv_proj.weight"):
                 # Transform checkpoint layout before entering vLLM's loader so
                 # online FP8 can keep ``online_process_loader`` outermost.
                 loaded_weight = _reorder_grouped_qkv_to_qkv(
@@ -1319,12 +1373,26 @@ class MiniMaxH3DiTModel(nn.Module):
                         "MiniMax H3 fc1 checkpoint rows must split evenly into "
                         f"gate/up matrices, got {tuple(loaded_weight.shape)}"
                     )
-                gate, up = loaded_weight.chunk(2, dim=0)
+                first, second = loaded_weight.chunk(2, dim=0)
+                gate, up = (second, first) if layout == "swap_halves" else (first, second)
                 weight_loader(param, gate, 0)
                 weight_loader(param, up, 1)
             else:
                 weight_loader(param, loaded_weight)
             loaded.add(name)
+        if diffusers_weights:
+            for name, parts in qkv_parts.items():
+                if parts != {"q", "k", "v"}:
+                    raise ValueError(f"incomplete Diffusers H3 QKV group {name}: {sorted(parts)}")
+            # Diffusers reconstructs RoPE from config instead of storing this
+            # native checkpoint buffer. Compute on CPU for identical values.
+            freq_dim = self.arch.rope_inv_freq_len
+            rope = 1.0 / (
+                self._rope_theta
+                ** (torch.arange(0, 2 * freq_dim, 2, dtype=torch.float32, device="cpu") / (2 * freq_dim))
+            )
+            default_weight_loader(params["rope.inv_freq"], rope)
+            loaded.add("rope.inv_freq")
         return loaded
 
     @staticmethod

vllm_omni/diffusion/models/minimax_h3/pipeline_minimax_h3.py

@@ -67,6 +67,7 @@ from vllm_omni.model_executor.model_loader.weight_utils import (
     download_weights_from_hf_specific,
 )
 from vllm_omni.model_executor.models.minimax_h3.checkpoint import (
+    is_minimax_h3_modular,
     resolve_minimax_h3_partition,
 )
 from vllm_omni.model_executor.models.minimax_h3.conditioning import (
@@ -106,6 +107,7 @@ from .denoise_loop import (
 )
 from .encoder import MiniMaxH3Qwen3VLEncoder
 from .fasth3 import FastH3WeightFusion, resolve_fasth3_fusion
+from .fasth3_checkpoint import FastH3CheckpointSpec
 from .lora import TurboSpec, load_minimax_h3_turbo_lora
 from .minimax_h3_transformer import (
     MiniMaxH3Attention,
@@ -213,7 +215,11 @@ def _resolve_minimax_h3_model_root(
         if path.name in {"FL2VA", "Ref2VA"}:
             return path.parent
         return path
-    if load_text_encoder:
+    if is_minimax_h3_modular(model, revision):
+        allow_patterns = ["modular_model_index.json", "fastvideo_inference.json", "provenance.json", "transformer/**"]
+        if load_text_encoder:
+            allow_patterns += ["text_encoder/**", "tokenizer/**", "processor/**"]
+    elif load_text_encoder:
         allow_patterns = (
             MINIMAX_H3_DOWNLOAD_PATTERNS if partition == "combined" else MINIMAX_H3_TASK_DOWNLOAD_PATTERNS[partition]
         )
@@ -314,6 +320,8 @@ def resolve_minimax_h3_diffusion_model_path(
         partition,
         load_text_encoder=False,
     )
+    if is_minimax_h3_modular(str(model_root), revision):
+        return str(model_root)
     if partition == "combined":
         return str(model_root)
     subdir = "Ref2VA" if partition == "ref2va" else "FL2VA"
@@ -575,6 +583,7 @@ class MiniMaxH3Pipeline(
     _base_schedule_by_partition: ClassVar[Mapping[str, DMD2SigmaSchedule | None]] = {}
     # Set from --lora-path during construction; absent means no FastH3 adapter.
     _fasth3: FastH3WeightFusion | None = None
+    _fasth3_checkpoint: FastH3CheckpointSpec | None = None
 
     def _load_diffusion_lora_adapter(
         self,
@@ -800,10 +809,13 @@ class MiniMaxH3Pipeline(
             self._PROFILER_TARGETS.remove("encode_prompt")
         if not self.load_vae_encoder:
             self._PROFILER_TARGETS.remove("_encode_local_media")
+        modular = is_minimax_h3_modular(str(od_config.model), od_config.revision)
         self.partition = _minimax_h3_partition_for_task(
             getattr(od_config, "task_type", None),
             str(od_config.model),
         )
+        if modular and str(od_config.task_type or "auto").lower() == "auto":
+            self.partition = "fl2va"
         self._turbo_lora_specs: dict[int, TurboSpec] = {}
         self._native_lora_adapter_ids: set[int] = set()
         self._lora_sigma_schedules: dict[int, DMD2SigmaSchedule] = {}
@@ -813,9 +825,19 @@ class MiniMaxH3Pipeline(
             self.partition,
             load_text_encoder=self.load_text_encoder,
         )
-        model_path = model_root / ("Ref2VA" if self.partition == "ref2va" else "FL2VA")
-        model_index = json.loads((model_path / "model_index.json").read_text(encoding="utf-8"))
-        release = model_index.get("_minimax_h3") or {}
+        if modular:
+            model_path = model_root
+            self._fasth3_checkpoint = FastH3CheckpointSpec.from_metadata(
+                json.loads((model_root / "fastvideo_inference.json").read_text(encoding="utf-8"))
+            )
+            self._fasth3_checkpoint.check_serving_contract(partition=self.partition, od_config=od_config)
+            release = self._fasth3_checkpoint.release_metadata()
+            vae_model_path = self._fasth3_checkpoint.resolve_native_vaes(model_root)
+        else:
+            model_path = model_root / ("Ref2VA" if self.partition == "ref2va" else "FL2VA")
+            model_index = json.loads((model_path / "model_index.json").read_text(encoding="utf-8"))
+            release = model_index.get("_minimax_h3") or {}
+            vae_model_path = model_path
         partition = str(release.get("partition", "")).lower()
         expected_partition = "ref2va" if self.partition == "ref2va" else "fl2va"
         if partition != expected_partition:
@@ -876,11 +898,18 @@ class MiniMaxH3Pipeline(
         self.transformer = MiniMaxH3DiTModel(
             od_config,
             quant_config=transformer_quant_config,
+            diffusers_weights=modular,
         )
+        if self._fasth3_checkpoint is not None:
+            self.transformer.enable_vsa_gates(sparsity=self._fasth3_checkpoint.vsa_sparsity)
+            logger.info(
+                "FastH3 V2 full checkpoint: 8 transformer forwards, video/audio shifts 10/3, VSA sparsity=0.8 tile=64"
+            )
         if ref2va_model_path is not None:
             self.transformers_ref = MiniMaxH3DiTModel(
                 od_config,
                 quant_config=transformer_quant_config,
+                diffusers_weights=modular,
             )
 
         self._fasth3 = resolve_fasth3_fusion(od_config, self.transformer)
@@ -963,14 +992,14 @@ class MiniMaxH3Pipeline(
         # so VAEs stay resident for new configurations.
         component_load_device = torch.device("cpu") if legacy_manual_components else self.device
         self.video_vae = MiniMaxH3VideoVAE(
-            os.path.join(model_path, "video_vae"),
+            os.path.join(vae_model_path, "video_vae"),
             device=self.device,
             load_device=component_load_device,
             decode_only=not self.load_vae_encoder,
             trust_remote_code=od_config.trust_remote_code,
         )
         self.audio_vae = MiniMaxH3AudioVAE(
-            os.path.join(model_path, "audio_vae"),
+            os.path.join(vae_model_path, "audio_vae"),
             device=self.device,
             load_device=component_load_device,
             decode_only=not self.load_vae_encoder,
@@ -1046,6 +1075,12 @@ class MiniMaxH3Pipeline(
             # load_weights only warns on a parameter the model does not have, so
             # close the adapter against what the DiT actually consumed.
             self._fasth3.validate_fully_applied(transformer_loaded)
+        if self._fasth3_checkpoint is not None:
+            required_gates = {
+                f"blocks.{i}.attn.to_gate_compress.weight" for i in range(self.transformer.arch.num_layers)
+            }
+            if missing_gates := required_gates - transformer_loaded:
+                raise ValueError(f"FastH3 V2 checkpoint is missing compression gates: {sorted(missing_gates)}")
         return loaded_with_prefix
 
     @property
@@ -2056,6 +2091,11 @@ class MiniMaxH3Pipeline(
             self._validate_turbo_sampling(sampling, turbo_spec)
         if has_native_lora:
             self._validate_native_sampling(sampling, task=task)
+        if self._fasth3_checkpoint is not None:
+            self._fasth3_checkpoint.check_request(
+                sampling,
+                step_execution=bool(getattr(self.od_config, "step_execution", False)),
+            )
         if self._fasth3 is not None:
             self._fasth3.check_request(
                 sampling,

vllm_omni/model_executor/models/minimax_h3/checkpoint.py

@@ -1,10 +1,16 @@
 # SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""MiniMax H3 checkpoint resolution for the AR text-encoder stage."""
+
 from pathlib import Path
 
+from vllm_omni.diffusion.utils.hf_utils import get_diffusion_model_index
 from vllm_omni.model_executor.model_loader.weight_utils import (
     download_weights_from_hf_specific,
 )
 
+_FASTH3_MODULAR_MODEL_PREFIX = "FastVideo/"
+
 MINIMAX_H3_ENCODER_DOWNLOAD_PATTERNS = {
     partition: [
         f"{subdir}/text_encoder/**",
@@ -16,6 +22,22 @@ MINIMAX_H3_ENCODER_DOWNLOAD_PATTERNS = {
 MINIMAX_H3_ENCODER_DOWNLOAD_PATTERNS["combined"] = MINIMAX_H3_ENCODER_DOWNLOAD_PATTERNS["fl2va"]
 
 
+def is_minimax_h3_modular(model: str, revision: str | None = None) -> bool:
+    path = Path(model)
+    if path.is_dir():
+        # Native MiniMax-H3 snapshots can advertise the Diffusers modular
+        # class in ``model_index.json`` while retaining the legacy FL2VA /
+        # Ref2VA layout.  FastH3 modular releases carry this marker file.
+        return (path / "modular_model_index.json").is_file() or (path / "fastvideo_inference.json").is_file()
+    # The FastVideo release is the modular checkpoint handled by this
+    # resolver.  Do not classify the native MiniMax-H3 Hub repository from
+    # its class name alone; its partition files are still required here.
+    if not model.startswith(_FASTH3_MODULAR_MODEL_PREFIX):
+        return False
+    index = get_diffusion_model_index(model, revision=revision) or {}
+    return index.get("_class_name") == "MiniMaxH3ModularPipeline"
+
+
 def resolve_minimax_h3_partition(
     model: str,
     task_type: str | None,
@@ -42,10 +64,13 @@ def resolve_minimax_h3_encoder_model_root(
 ) -> str:
     path = Path(model)
     partition = resolve_minimax_h3_partition(model, task_type, auto_partition="fl2va")
+    modular = is_minimax_h3_modular(model, revision)
 
     if path.is_dir():
         if path.name == "text_encoder" and (path / "config.json").is_file():
             return str(path)
+        if modular:
+            return str(path / "text_encoder")
         if path.name in {"FL2VA", "Ref2VA"}:
             path = path.parent
         subdir = "Ref2VA" if partition == "ref2va" else "FL2VA"
@@ -53,11 +78,13 @@ def resolve_minimax_h3_encoder_model_root(
     snapshot = download_weights_from_hf_specific(
         model_name_or_path=model,
         cache_dir=None,
-        allow_patterns=MINIMAX_H3_ENCODER_DOWNLOAD_PATTERNS[partition],
+        allow_patterns=["text_encoder/**"] if modular else MINIMAX_H3_ENCODER_DOWNLOAD_PATTERNS[partition],
         revision=revision,
         require_all=True,
     )
 
+    if modular:
+        return str(Path(snapshot) / "text_encoder")
     subdir = "Ref2VA" if partition == "ref2va" else "FL2VA"
     return str(Path(snapshot) / subdir / "text_encoder")
 

vllm_omni/platforms/npu/worker/npu_ar_model_runner.py

@@ -177,6 +177,7 @@ class NPUARModelRunner(OmniNPUModelRunner, OmniConnectorModelRunnerMixin, Duplex
                         num_scheduled_tokens_np=np.ones(bsz, dtype=np.int32),
                         max_num_scheduled_tokens=1,
                         use_cascade_attn=False,
+                        force_uniform_decode=True,
                     )
                     n = batch_desc.num_tokens
                     ids = self.talker_mtp_input_ids.gpu[:n]