Merge main and adapt multi-API startup to structured configs

Signed-off-by: Sy03 <1370724210@qq.com>
Merge main and preserve multi-API serving mode selection
2026-09-17 12:24:06 +08:00 · 2026-09-17 11:49:47 +08:00 · 2026-09-16 21:02:33 +08:00 · 2026-09-16 20:18:31 +08:00 · 2026-09-16 16:40:16 +08:00 · 2026-09-16 16:00:45 +08:00
18 changed files with 113 additions and 2307 deletions
@@ -271,111 +271,3 @@ 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.
@@ -1106,68 +1106,6 @@ 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 |
@@ -1,679 +0,0 @@
-# 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))
@@ -423,7 +423,6 @@ 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")]
@@ -262,10 +262,7 @@ 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.not_a_projection.lora_A.weight": torch.ones((_RANK, _HIDDEN))},
-    )
+    _write_adapter(path, tensors={"transformer_blocks.0.attn.norm_q.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)
@@ -10,7 +10,6 @@ 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_,
@@ -191,55 +190,3 @@ 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)
@@ -1,5 +1,4 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project

 from types import SimpleNamespace
 from unittest.mock import Mock
@@ -42,7 +41,7 @@ class _FakeAttention(nn.Module):
        super().__init__()


-def _small_od_config(*, class_name=None):
+def _small_od_config():
    arch = {
        "num_layers": 1,
        "token_refiner_num_layers": 1,
@@ -61,8 +60,6 @@ def _small_od_config(*, class_name=None):
        "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),
@@ -136,24 +133,6 @@ 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__()
@@ -187,11 +187,7 @@ 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.
-        # 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,
+        "stage_0_tensor_parallel_size": 2,
        "vae_use_tiling": True,
        # MiniMax H3's VAE is implemented by code shipped with the checkpoint.
        "trust_remote_code": True,
@@ -6,7 +6,6 @@ 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"
@@ -22,19 +21,17 @@ 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"
@@ -10,24 +10,24 @@ import base64
 import hashlib
 import json
 import math
-import time
 import uuid
 import wave
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import 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, WebSocketException
+    from websockets.exceptions import ConnectionClosed
 except ImportError as exc:  # pragma: no cover - driver dependency.
    raise SystemExit("Install websockets first: pip install websockets") from exc

-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 (
+    EventCollector,
+    write_pcm16_wav,
+)
 from vllm_omni.clients.duplex import (
    wait_for_condition as wait_for,
 )
@@ -35,24 +35,6 @@ 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:
@@ -64,33 +46,29 @@ class RawRealtimeProbe:
    keeps a transport-only client and reads everything from ``.events``.
    """

-    def __init__(self, url: str, *, max_size: int = 64 * 1024 * 1024, close_timeout_s: float = 10.0) -> None:
+    def __init__(self, url: str, *, max_size: int = 64 * 1024 * 1024) -> None:
        self.url = url
        self.max_size = max_size
-        self.close_timeout_s = close_timeout_s
        self.events = EventCollector()
-        self._ws: _ProbeTransport | None = None
+        self._ws = 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, close_timeout=self.close_timeout_s)
+        self._ws = await websockets.connect(self.url, max_size=self.max_size)
        self._reader_task = asyncio.create_task(self._read_events())
        return self

    async def __aexit__(self, exc_type, exc, traceback) -> None:
-        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
+        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

    async def _read_events(self) -> None:
-        assert self._ws is not None
        try:
            while True:
                raw = await self._ws.recv()
@@ -103,7 +81,6 @@ 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))


@@ -165,38 +142,6 @@ 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,
    *,
@@ -206,7 +151,22 @@ 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(_session_update(args, session_id=session_id, persona=persona))
+    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},
+            },
+        }
+    )
    event_type = "error" if expect_error else "session.created"
    await wait_for(
        lambda: client.events.count(event_type) > 0,
@@ -216,23 +176,6 @@ 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(
@@ -257,24 +200,23 @@ 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]
-        await client.send(_frame_event(frame, seq))
+        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 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,
    *,
@@ -300,7 +242,11 @@ def _audio_frame_stats(


 def _response_ids(client: RawRealtimeProbe) -> set[str]:
-    return {str(event["response_id"]) for event in _audio_events(client) if isinstance(event.get("response_id"), str)}
+    return {
+        str(event["response_id"])
+        for event in _events(client, "response.output_audio.delta")
+        if isinstance(event.get("response_id"), str)
+    }


 def _session_result(
@@ -310,66 +256,40 @@ def _session_result(
    args: argparse.Namespace,
    minimum_chunks: int,
 ) -> tuple[bytes, set[str], dict[str, object]]:
-    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)
+    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)}")
    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 _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)}
+        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)
+    }
    if rates != {SAMPLE_RATE_HZ}:
-        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,
-        )
+        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}")
    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(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,
+    if int(stats["voiced_frames"]) < min_voiced_frames:
+        raise AssertionError(
+            f"output contained insufficient audible speech: minimum_voiced_frames={min_voiced_frames}, stats={stats}"
        )
    runtime_metadata = [
        metadata["vllm_omni"]
-        for event in _audio_events(client)
+        for event in _events(client, "response.output_audio.delta")
        if isinstance((metadata := event.get("metadata")), dict) and isinstance(metadata.get("vllm_omni"), dict)
    ]
    if not any(
@@ -378,19 +298,18 @@ def _session_result(
        and item.get("runner_kv_backed") is True
        for item in runtime_metadata
    ):
-        raise _AudioAcceptanceError(
-            "scheduler_data_plane", "audio events did not prove the scheduler data plane", stats
-        )
+        raise AssertionError(f"audio events did not prove the scheduler data plane: {runtime_metadata}")
    response_ids = _response_ids(client)
    if len(response_ids) != 1:
-        raise _AudioAcceptanceError(
-            "response_identity", f"continuous stream requires one response: count={len(response_ids)}", stats
-        )
+        raise AssertionError(f"continuous stream used multiple responses: {sorted(response_ids)}")
    return (
        raw,
        response_ids,
        {
            **stats,
+            "audio_chunks": chunks,
+            "audio_samples": len(raw) // 2,
+            "audio_rms": rms,
            "response_ids": sorted(response_ids),
        },
    )
@@ -413,8 +332,6 @@ 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,
@@ -464,12 +381,14 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
    )
    await asyncio.gather(
        wait_for(
-            lambda: len(_audio_bytes(primary)) // (2 * FRAME_SAMPLES) >= primary_frames - args.max_frame_deficit,
+            lambda: len(primary.events.audio_bytes()) // (2 * FRAME_SAMPLES) >= primary_frames - args.max_frame_deficit,
            timeout_s=args.timeout_s,
            label="primary frame coverage",
        ),
        wait_for(
-            lambda: (len(_audio_bytes(secondary)) // (2 * FRAME_SAMPLES) >= secondary_frames - args.max_frame_deficit),
+            lambda: (
+                len(secondary.events.audio_bytes()) // (2 * FRAME_SAMPLES) >= secondary_frames - args.max_frame_deficit
+            ),
            timeout_s=args.timeout_s,
            label="secondary frame coverage",
        ),
@@ -507,14 +426,16 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
    await asyncio.gather(
        wait_for(
            lambda: (
-                len(_audio_bytes(secondary)) // (2 * FRAME_SAMPLES) >= secondary_total_frames - args.max_frame_deficit
+                len(secondary.events.audio_bytes()) // (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(_audio_bytes(replacement)) // (2 * FRAME_SAMPLES) >= replacement_frames - args.max_frame_deficit
+                len(replacement.events.audio_bytes()) // (2 * FRAME_SAMPLES)
+                >= replacement_frames - args.max_frame_deficit
            ),
            timeout_s=args.timeout_s,
            label="replacement frame coverage",
@@ -572,284 +493,6 @@ 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")
@@ -867,12 +510,6 @@ 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,
@@ -899,39 +536,11 @@ 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.",
    )
-    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
+    return parser.parse_args(argv)


 def main() -> None:
-    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)
+    print(json.dumps(asyncio.run(run(parse_args())), ensure_ascii=False, indent=2))


 if __name__ == "__main__":
@@ -1,568 +0,0 @@
-# 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)
@@ -24,18 +24,6 @@ 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,
@@ -114,7 +102,6 @@ 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
@@ -125,20 +112,13 @@ 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
-    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
+    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)
    tl.store(
        output_ptr + row * hidden_size + columns,
-        output,
+        normalized * (1.0 + scale) + shift,
        mask=mask,
    )

@@ -163,7 +143,6 @@ 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
@@ -180,20 +159,12 @@ 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
-    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
+    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)
    tl.store(
        modulated_out_ptr + row * hidden_size + columns,
-        output,
+        normalized * (1.0 + scale) + shift,
        mask=mask,
    )

@@ -300,7 +271,6 @@ 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,
        )
@@ -354,7 +324,6 @@ 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,
        )
@@ -102,7 +102,6 @@ _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
@@ -116,8 +115,6 @@ _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),
@@ -1,131 +0,0 @@
-# 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")
@@ -55,8 +55,6 @@ 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,
@@ -120,18 +118,6 @@ 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:
@@ -408,9 +394,6 @@ 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,
@@ -605,7 +588,6 @@ 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 {}),
@@ -1145,23 +1127,10 @@ 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
@@ -1242,7 +1211,7 @@ class MiniMaxH3DiTModel(nn.Module):
        )
        self._mark_missing_params_required()

-    def enable_vsa_gates(self, *, sparsity: float | None = None) -> None:
+    def enable_vsa_gates(self) -> None:
        """Give every DiT block's attention a VSA compression gate.

        A FastH3 VSA artifact assigns these projections rather than adding to
@@ -1254,9 +1223,6 @@ 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:
@@ -1327,37 +1293,17 @@ class MiniMaxH3DiTModel(nn.Module):
        self,
        weights: Iterable[tuple[str, torch.Tensor]],
    ) -> set[str]:
-        """Load native or Diffusers H3 weights with the existing TP-aware loaders."""
+        """Load exact H3 checkpoint names with logical 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 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"):
+            if 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(
@@ -1373,26 +1319,12 @@ class MiniMaxH3DiTModel(nn.Module):
                        "MiniMax H3 fc1 checkpoint rows must split evenly into "
                        f"gate/up matrices, got {tuple(loaded_weight.shape)}"
                    )
-                first, second = loaded_weight.chunk(2, dim=0)
-                gate, up = (second, first) if layout == "swap_halves" else (first, second)
+                gate, up = loaded_weight.chunk(2, dim=0)
                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
@@ -67,7 +67,6 @@ 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 (
@@ -107,7 +106,6 @@ 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,
@@ -215,11 +213,7 @@ def _resolve_minimax_h3_model_root(
        if path.name in {"FL2VA", "Ref2VA"}:
            return path.parent
        return path
-    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:
+    if load_text_encoder:
        allow_patterns = (
            MINIMAX_H3_DOWNLOAD_PATTERNS if partition == "combined" else MINIMAX_H3_TASK_DOWNLOAD_PATTERNS[partition]
        )
@@ -320,8 +314,6 @@ 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"
@@ -583,7 +575,6 @@ 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,
@@ -809,13 +800,10 @@ 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] = {}
@@ -825,19 +813,9 @@ class MiniMaxH3Pipeline(
            self.partition,
            load_text_encoder=self.load_text_encoder,
        )
-        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
+        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 {}
        partition = str(release.get("partition", "")).lower()
        expected_partition = "ref2va" if self.partition == "ref2va" else "fl2va"
        if partition != expected_partition:
@@ -898,18 +876,11 @@ 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)
@@ -992,14 +963,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(vae_model_path, "video_vae"),
+            os.path.join(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(vae_model_path, "audio_vae"),
+            os.path.join(model_path, "audio_vae"),
            device=self.device,
            load_device=component_load_device,
            decode_only=not self.load_vae_encoder,
@@ -1075,12 +1046,6 @@ 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
@@ -2091,11 +2056,6 @@ 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,
@@ -1,16 +1,10 @@
 # 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/**",
@@ -22,22 +16,6 @@ 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,
@@ -64,13 +42,10 @@ 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"
@@ -78,13 +53,11 @@ def resolve_minimax_h3_encoder_model_root(
    snapshot = download_weights_from_hf_specific(
        model_name_or_path=model,
        cache_dir=None,
-        allow_patterns=["text_encoder/**"] if modular else MINIMAX_H3_ENCODER_DOWNLOAD_PATTERNS[partition],
+        allow_patterns=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")

@@ -177,7 +177,6 @@ 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]
Author	SHA1	Message	Date
Sy03	04ae6aa009	Merge main and adapt multi-API startup to structured configs Signed-off-by: Sy03 <1370724210@qq.com>	2026-09-17 12:24:06 +08:00
Sy03	79c629885a	Merge main and preserve multi-API serving mode selection Signed-off-by: Sy03 <1370724210@qq.com>	2026-09-17 11:49:47 +08:00
Sy03	d8458b8417	Revert unrelated ROCm CI test and timeout changes Signed-off-by: Sy03 <1370724210@qq.com>	2026-09-16 21:02:33 +08:00
Sy03	ec0feea2bd	Fix ROCm diffusion CI platform guard and startup deadlines Signed-off-by: Sy03 <1370724210@qq.com>	2026-09-16 20:18:31 +08:00
Sy03	daf91b4d7a	Merge main and preserve multi-API engine and duplex guards Signed-off-by: Sy03 <1370724210@qq.com>	2026-09-16 16:40:16 +08:00
Sy03	40a000de05	Document multi-API serving tradeoffs in TTS profile Signed-off-by: Sy03 <1370724210@qq.com>	2026-09-16 16:00:45 +08:00
Sy03	dc3c7dbe16	Rebase multi-API serving on latest main Signed-off-by: Sy03 <1370724210@qq.com>	2026-09-16 12:13:04 +08:00