Merge commit '6fe1ff7709c72c08bbfc33a7e14b73ec94135549' into codex-issue-7035-h100-determinism

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

tests/diffusion/models/minimax_h3/test_fasth3_checkpoint.py

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

tests/diffusion/models/minimax_h3/test_minimax_h3_contract.py

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

tests/diffusion/models/minimax_h3/test_minimax_h3_fasth3.py

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

tests/diffusion/models/minimax_h3/test_minimax_h3_quantization.py

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

tests/diffusion/models/minimax_h3/test_minimax_h3_quantization_quality.py

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

tests/diffusion/models/wan2_2/test_create_transformer_quant.py

@@ -1,5 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
 """Regression tests for transformer quant-config auto-detection.
 
 The loader path at pipeline_wan2_2.py carries two quantization contracts:
@@ -27,6 +27,7 @@ from types import SimpleNamespace
 import pytest
 
 import vllm_omni.diffusion.models.wan2_2.pipeline_wan2_2 as wan22_module
+from vllm_omni.diffusion.data import OmniDiffusionConfig
 from vllm_omni.diffusion.models.wan2_2.pipeline_wan2_2 import create_transformer_from_config
 
 pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu]
@@ -381,3 +382,213 @@ def test_pipeline_cascade_mxfp4_dualscale_each_transformer_gets_correct_ignored_
 
     # od_config must remain unchanged — _create_transformer does not modify it.
     assert od_config.quantization_config is None
+
+
+def _construct_quantized_pipeline(pipeline_kind, active, disk_quant_config, tmp_path, monkeypatch):
+    import json
+
+    import torch
+
+    import vllm_omni.diffusion.models.wan2_2.pipeline_wan2_2_i2v as i2v_module
+    import vllm_omni.diffusion.models.wan2_2.pipeline_wan2_2_vace as vace_module
+
+    monkeypatch.setattr("vllm_omni.platforms.current_omni_platform.is_npu", lambda: True)
+    for component in ("transformer", "transformer_2"):
+        directory = tmp_path / component
+        directory.mkdir()
+        config = dict(_MIN_CFG)
+        expert_quant_config = (
+            disk_quant_config[0 if component == "transformer" else 1]
+            if isinstance(disk_quant_config, tuple)
+            else disk_quant_config
+        )
+        if expert_quant_config is not None:
+            config["quantization_config"] = expert_quant_config
+        (directory / "config.json").write_text(json.dumps(config))
+    (tmp_path / "model_index.json").write_text(json.dumps({"transformer_2": ["diffusers", "WanTransformer3DModel"]}))
+
+    class FakeComponent(torch.nn.Module):
+        def __init__(self):
+            super().__init__()
+            self.config = SimpleNamespace(scale_factor_temporal=4, scale_factor_spatial=8)
+
+    class FakeTransformer(FakeComponent):
+        def __init__(self, quant_config=None, **kwargs):
+            super().__init__()
+            self.received_quant_config = quant_config
+
+    for module in (wan22_module, i2v_module):
+        monkeypatch.setattr(module, "get_local_device", lambda: torch.device("cpu"))
+        monkeypatch.setattr(module, "prefetch_subfolders", lambda *args, **kwargs: None)
+        monkeypatch.setattr(module, "from_pretrained_with_prefetch", lambda *args, **kwargs: FakeComponent())
+    monkeypatch.setattr(wan22_module, "WanTransformer3DModel", FakeTransformer)
+    monkeypatch.setattr(vace_module, "WanVACETransformer3DModel", FakeTransformer)
+    pipeline_cls = {
+        "t2v": wan22_module.Wan22Pipeline,
+        "i2v": i2v_module.Wan22I2VPipeline,
+        "vace": vace_module.Wan22VACEPipeline,
+    }[pipeline_kind]
+    monkeypatch.setattr(pipeline_cls, "setup_diffusion_pipeline_profiler", lambda *args, **kwargs: None)
+    od_config = OmniDiffusionConfig(
+        model=str(tmp_path),
+        model_class_name=pipeline_cls.__name__,
+        dtype=torch.bfloat16,
+        quantization_config=active,
+        boundary_ratio=0.875,
+        flow_shift=12.0,
+        enable_diffusion_pipeline_profiler=False,
+    )
+    od_config.enrich_config()
+    pipeline = pipeline_cls(od_config=od_config)
+    return pipeline
+
+
+@pytest.mark.parametrize("pipeline_kind", ["t2v", "i2v", "vace"])
+@pytest.mark.parametrize("method, flag", [("mxfp4", "is_checkpoint_mxfp4_serialized")])
+@pytest.mark.parametrize("offline", [False, True])
+def test_pipeline_constructor_routes_each_expert_quantization(
+    pipeline_kind, method, flag, offline, tmp_path, monkeypatch, mocker
+):
+    """Run the actual pipeline constructors, expert factories and quant-method dispatch."""
+    from vllm.model_executor.layers.linear import LinearBase, UnquantizedLinearMethod
+
+    from vllm_omni.quantization import build_quant_config
+
+    active = build_quant_config(
+        {
+            "transformer": {
+                "method": method,
+                "w4a8_fallback_layers": ["blocks.10.attn1.to_qkv"],
+                "w4a8_fallback_steps": [0, 1],
+            },
+            "transformer_2": {
+                "method": method,
+                "w4a8_fallback_layers": ["blocks.11.attn1.to_qkv"],
+                "w4a8_fallback_steps": [],
+            },
+        }
+    )
+    disk_quant_config = None
+    if offline:
+        disk_quant_config = {
+            "quant_method": method,
+            flag: True,
+            "ignored_layers": ["blocks.12.ffn.net_2"],
+            "w4a8_fallback_layers": ["blocks.9.attn1.to_qkv"],
+            "w4a8_fallback_steps": [9],
+        }
+    pipeline = _construct_quantized_pipeline(pipeline_kind, active, disk_quant_config, tmp_path, monkeypatch)
+    for index, transformer in enumerate((pipeline.transformer, pipeline.transformer_2)):
+        received = transformer.received_quant_config
+        assert received.get_name() == method
+        assert getattr(received, flag) == offline
+        assert received.w4a8_fallback_steps == ([0, 1] if index == 0 else [])
+        selected = f"blocks.{10 + index}.attn1.to_qkv"
+        other = f"blocks.{11 - index}.attn1.to_qkv"
+        assert received.w4a8_fallback_layers == [selected]
+        assert received.get_quant_method(mocker.Mock(spec=LinearBase), selected).is_w4a8_fallback_layer
+        assert not received.get_quant_method(mocker.Mock(spec=LinearBase), other).is_w4a8_fallback_layer
+        if offline:
+            assert isinstance(
+                received.get_quant_method(mocker.Mock(spec=LinearBase), "blocks.12.ffn.net_2"), UnquantizedLinearMethod
+            )
+    assert pipeline.od_config.quantization_config is active
+
+
+@pytest.mark.parametrize("explicit_none", [False, True])
+def test_pipeline_disabled_component_requires_unquantized_checkpoint(explicit_none, monkeypatch):
+    from vllm_omni.quantization import build_quant_config
+
+    FakeTransformer, captured = _make_fake_transformer()
+    monkeypatch.setattr(wan22_module, "WanTransformer3DModel", FakeTransformer)
+    components = {"transformer": {"method": "mxfp4"}, "transformer_2": None} if explicit_none else {"vae": None}
+    active = build_quant_config(components)
+    pipeline = _FakePipeline(OmniDiffusionConfig(model="", quantization_config=active))
+    pipeline._create_transformer(_MIN_CFG, component="transformer_2")
+    assert captured[-1].get("quant_config") is None
+    with pytest.raises(ValueError, match="disabled for component 'transformer_2'"):
+        pipeline._create_transformer(
+            {**_MIN_CFG, "quantization_config": {"quant_method": "mxfp4", "is_checkpoint_mxfp4_serialized": True}},
+            component="transformer_2",
+        )
+
+
+def test_pipeline_component_default_reaches_expert_factory(monkeypatch):
+    from vllm_omni.quantization import build_quant_config
+
+    FakeTransformer, captured = _make_fake_transformer()
+    monkeypatch.setattr(wan22_module, "WanTransformer3DModel", FakeTransformer)
+    active = build_quant_config({"default": {"method": "mxfp4", "w4a8_fallback_layers": ["blocks.10.attn1.to_qkv"]}})
+    pipeline = _FakePipeline(OmniDiffusionConfig(model="", quantization_config=active))
+    pipeline._create_transformer(_MIN_CFG, component="transformer_2")
+    assert captured[-1]["quant_config"] is active.default_config
+
+
+@pytest.mark.parametrize("pipeline_kind", ["t2v", "i2v", "vace"])
+@pytest.mark.parametrize(
+    "configured_expert, with_default",
+    [
+        ("transformer", False),
+        ("transformer", True),
+        ("transformer_2", False),
+        ("transformer_2", True),
+    ],
+)
+def test_pipeline_partial_expert_config_does_not_match_sibling(
+    pipeline_kind, configured_expert, with_default, tmp_path, monkeypatch
+):
+    from vllm_omni.quantization import build_quant_config
+
+    component_spec = {
+        configured_expert: {"method": "mxfp4", "w4a8_fallback_layers": ["blocks.10.attn1.to_qkv"]},
+    }
+    if with_default:
+        component_spec["default"] = {"method": "mxfp4", "w4a8_fallback_layers": ["blocks.11.attn1.to_qkv"]}
+    active = build_quant_config(component_spec)
+    pipeline = _construct_quantized_pipeline(pipeline_kind, active, None, tmp_path, monkeypatch)
+    sibling = "transformer_2" if configured_expert == "transformer" else "transformer"
+    selected = getattr(pipeline, configured_expert).received_quant_config
+    other = getattr(pipeline, sibling).received_quant_config
+    assert selected is active.component_configs[configured_expert]
+    assert other is active.default_config
+    if with_default:
+        assert other.w4a8_fallback_layers == ["blocks.11.attn1.to_qkv"]
+    else:
+        assert other is None
+
+
+@pytest.mark.parametrize("pipeline_kind", ["t2v", "i2v", "vace"])
+@pytest.mark.parametrize("different_ignored_layers", [False, True])
+def test_enriched_auto_quantization_preserves_each_expert_policy(
+    pipeline_kind, different_ignored_layers, tmp_path, monkeypatch
+):
+    """Startup auto-detection must not turn the first expert into a user override."""
+    high = {
+        "quant_method": "mxfp4",
+        "is_checkpoint_mxfp4_serialized": True,
+        "ignored_layers": ["blocks.0.ffn.net_2"],
+        "w4a8_fallback_steps": [],
+        "w4a8_fallback_layers": [],
+        "mxfp4_scale_alg": 0,
+        "require_smooth_scale": False,
+    }
+    low = {
+        **high,
+        "ignored_layers": ["blocks.1.ffn.net_2"] if different_ignored_layers else high["ignored_layers"],
+        "w4a8_fallback_steps": [37],
+        "w4a8_fallback_layers": ["blocks.11.attn1.to_qkv"],
+        "mxfp4_scale_alg": 2,
+        "require_smooth_scale": True,
+    }
+    pipeline = _construct_quantized_pipeline(pipeline_kind, None, (high, low), tmp_path, monkeypatch)
+    assert pipeline.od_config.quantization_config is not None  # enrich_config really detected the first expert.
+    for transformer, expected in [(pipeline.transformer, high), (pipeline.transformer_2, low)]:
+        config = transformer.received_quant_config
+        for policy in (
+            "ignored_layers",
+            "w4a8_fallback_steps",
+            "w4a8_fallback_layers",
+            "mxfp4_scale_alg",
+            "require_smooth_scale",
+        ):
+            assert getattr(config, policy) == expected[policy], policy

tests/diffusion/models/wan2_2/test_wan22_pipeline_diffuse.py

@@ -20,9 +20,11 @@ pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.diffusion]
 
 
 @pytest.fixture(autouse=True)
-def _cpu_platform(monkeypatch):
+def _cpu_pipeline_runtime(monkeypatch):
+    # These lightweight pipelines have no accelerator or distributed groups.
     module = importlib.import_module("vllm_omni.diffusion.models.wan2_2.pipeline_wan2_2")
     monkeypatch.setattr(module.current_omni_platform, "is_available", lambda: False)
+    monkeypatch.setattr("vllm_omni.diffusion.distributed.parallel_state._PP", SimpleNamespace(world_size=1))
 
 
 class _StubTransformer(nn.Module):
@@ -477,6 +479,100 @@ def test_diffuse_runs_prediction_and_scheduler_for_each_timestep() -> None:
     assert torch.equal(result, torch.full_like(latents, 10.0))
 
 
+@pytest.mark.parametrize("fail_first_request", [False, True])
+def test_runner_publishes_wan_step_context_across_requests(fail_first_request, monkeypatch):
+    from vllm_omni.diffusion.data import OmniDiffusionConfig
+    from vllm_omni.diffusion.forward_context import (
+        ForwardContext,
+        get_forward_context,
+        is_forward_context_available,
+        override_forward_context,
+    )
+    from vllm_omni.diffusion.worker.diffusion_model_runner import DiffusionModelRunner
+    from vllm_omni.quantization.mxfp4_config import _is_w4a8_fallback_step
+
+    contexts: list[ForwardContext] = []
+    events = []
+
+    class InjectedPredictionError(Exception):
+        pass
+
+    class RecordingTransformer(_StubTransformer):
+        def __init__(self, expert):
+            super().__init__()
+            self.expert = expert
+
+        def forward(self, hidden_states, encoder_hidden_states, **kwargs):
+            context = get_forward_context()
+            assert context is contexts[-1]
+            request_index = len(contexts) - 1
+            step = context.denoise_step_idx
+            positive = bool(encoder_hidden_states[0, 0, 0] > 0)
+            events.append((request_index, step, self.expert, positive, _is_w4a8_fallback_step([0, 2])))
+            if fail_first_request and request_index == 0 and step == 1 and not positive:
+                raise InjectedPredictionError
+            return (torch.ones_like(hidden_states) * (1 if positive else -1),)
+
+    def encode_prompt(**kwargs):
+        context = get_forward_context()
+        assert context.denoise_step_idx is None
+        contexts.append(context)
+        embeds = torch.ones(1, 8, 8)
+        return embeds, -embeds
+
+    pipeline = _make_pipeline()
+    pipeline.transformer = RecordingTransformer("high")
+    pipeline.transformer_2 = RecordingTransformer("low")
+    pipeline.scheduler = _StubScheduler([900, 500, 100])
+    monkeypatch.setattr(pipeline, "encode_prompt", encode_prompt)
+    monkeypatch.setattr(pipeline, "scheduler_step_maybe_with_cfg", lambda pred, t, latents, cfg: latents)
+
+    # Use the real runner context manager, Wan forward/diffuse, and CFG
+    # dispatch. Only model computation, scheduler math and unrelated KV I/O
+    # are replaced; no test code publishes a denoise step.
+    runner = object.__new__(DiffusionModelRunner)
+    runner.pipeline = pipeline
+    runner.od_config = OmniDiffusionConfig(model="", dtype=torch.float32)
+    runner.vllm_config = None
+    runner.cache_backend = None
+    monkeypatch.setattr(runner, "_prepare_request_for_forward", lambda *args, **kwargs: None)
+    with override_forward_context(None):
+        for request_index in range(2):
+            request = OmniDiffusionRequest(
+                prompt="a fox walks",
+                request_id=f"request-{request_index}",
+                sampling_params=OmniDiffusionSamplingParams(
+                    num_frames=1, num_inference_steps=3, guidance_scale=2.0, output_type="latent"
+                ),
+            )
+
+            def execute():
+                return runner._execute_request_list(
+                    [request],
+                    od_config=runner.od_config,
+                    allow_single_output=True,
+                    require_request_batch_support=False,
+                    record_name="test_wan_step_context",
+                    record_output_peak_memory=False,
+                )
+
+            if fail_first_request and request_index == 0:
+                with pytest.raises(InjectedPredictionError):
+                    execute()
+            else:
+                execute()
+            assert not is_forward_context_available()
+
+    assert len(contexts) == 2 and contexts[0] is not contexts[1]
+    expected = [
+        (request, step, "high" if step == 0 else "low", positive, step in (0, 2))
+        for request in range(2)
+        for step in range(2 if fail_first_request and request == 0 else 3)
+        for positive in (True, False)
+    ]
+    assert events == expected
+
+
 class _StubDMDScheduler:
     def __init__(self) -> None:
         self.predict_clean_calls: list[tuple[float, float, float]] = []