[Feature][MiniMax-H3] Support the full LightX2V Turbo LoRA matrix (#7062)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

tests/diffusion/models/minimax_h3/test_minimax_h3_contract.py

@@ -13,6 +13,8 @@ import pytest
 import torch
 import torch.nn as nn
 
+from vllm_omni.diffusion.models.minimax_h3.lora import TurboSpec, parse_turbo_filename
+
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu, pytest.mark.diffusion]
 
 _ItemT = TypeVar("_ItemT")
@@ -24,6 +26,13 @@ def _append_and_return(items: list[_ItemT], item: _ItemT, result: _ResultT) -> _
     return result
 
 
+def _turbo_spec(filename: str) -> TurboSpec:
+    """Build the spec a loaded artifact of this name would carry."""
+    spec = parse_turbo_filename(filename)
+    assert spec is not None
+    return spec
+
+
 def test_h3_prepares_resolved_cache_state_immediately_before_denoise():
     from vllm_omni.diffusion.models.minimax_h3 import MiniMaxH3Pipeline
     from vllm_omni.diffusion.request import OmniDiffusionRequest
@@ -244,9 +253,12 @@ def test_combined_task_inference_and_transformer_routing():
     assert pipeline._resolve_task(None, {"image": object()}) == "fl2va"
     assert pipeline._resolve_task(None, {"audio": object()}) == "ref2va"
     assert pipeline._resolve_task(None, {"video": object()}) == "ref2va"
-    assert pipeline._resolve_task("fl2va", {"image": object()}, has_turbo_lora=True) == "fl2va"
-    with pytest.raises(OmniClientError, match="supports T2VA/FL2VA requests only"):
-        pipeline._resolve_task("ref2va", {}, has_turbo_lora=True)
+    fl2v_spec = _turbo_spec("minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors")
+    ref2v_spec = _turbo_spec("minimax_h3_ref2v_turbo_8step_v1.0_768p_bf16.safetensors")
+    assert pipeline._resolve_task("fl2va", {"image": object()}, turbo_spec=fl2v_spec) == "fl2va"
+    with pytest.raises(OmniClientError, match="serves"):
+        pipeline._resolve_task("ref2va", {}, turbo_spec=fl2v_spec)
+    assert pipeline._resolve_task("ref2va", {"image": object()}, turbo_spec=ref2v_spec) == "ref2va"
     pipeline.partition = "ref2va"
     pipeline.supported_tasks = frozenset({"ref2va"})
     assert pipeline._resolve_task(None, {"image": object()}) == "ref2va"

tests/diffusion/models/minimax_h3/test_minimax_h3_fasth3.py

@@ -303,7 +303,7 @@ def _pipeline_stub(fusion=None):
     torch.nn.Module.__init__(pipeline)
     pipeline.partition = "combined"
     pipeline.supported_tasks = frozenset({"t2va", "fl2va", "ref2va"})
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._fasth3 = fusion
     pipeline.od_config = SimpleNamespace()
     pipeline.default_video_shift, pipeline.default_audio_shift = 12.0, 3.0

tests/diffusion/models/minimax_h3/test_minimax_h3_lora.py

@@ -13,7 +13,9 @@ from vllm.lora.lora_weights import PackedLoRALayerWeights
 from vllm_omni.diffusion.lora.manager import DiffusionLoRAManager
 from vllm_omni.diffusion.models.minimax_h3 import lora as lora_module
 from vllm_omni.diffusion.models.minimax_h3.lora import (
+    TurboSpec,
     load_minimax_h3_turbo_lora,
+    parse_turbo_filename,
 )
 from vllm_omni.errors import OmniClientError
 from vllm_omni.lora.request import LoRARequest
@@ -46,7 +48,7 @@ def _request(path) -> LoRARequest:
 def _write_tiny_turbo(
     path,
     *,
-    alpha: str = "128",
+    alpha: str | None = "128",
     key_format: str = "minimax-h3-diffusers",
     omit_target: str | None = None,
     shape_overrides: dict[str, tuple[int, int]] | None = None,
@@ -88,11 +90,17 @@ def _write_tiny_turbo(
                         )
                 else:
                     tensors[b_name] = torch.ones(overrides.get(b_name, (output_dim, rank)))
-    save_file(
-        tensors,
-        str(path),
-        metadata={"alpha": alpha, "key_format": key_format},
-    )
+    metadata = {"key_format": key_format}
+    if alpha is not None:
+        metadata["alpha"] = alpha
+    save_file(tensors, str(path), metadata=metadata)
+
+
+def _spec(filename: str = "minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors") -> TurboSpec:
+    """Build the spec a loaded artifact of this name would carry."""
+    spec = parse_turbo_filename(filename)
+    assert spec is not None
+    return spec
 
 
 def test_h3_turbo_loads_through_legacy_lora_model_and_swaps_ffn(tmp_path):
@@ -107,9 +115,12 @@ def test_h3_turbo_loads_through_legacy_lora_model_and_swaps_ffn(tmp_path):
     )
 
     assert loaded is not None
-    lora_model, peft_helper = loaded
+    lora_model, peft_helper, spec = loaded
     assert peft_helper.r == 128
     assert peft_helper.lora_alpha == 128
+    assert spec.denoise_steps == 4
+    assert spec.sigma_points == 5
+    assert spec.task_family == "fl2v"
     assert len(lora_model.loras) == 312
     assert "blocks.0.attn.to_q" in lora_model.loras
     assert "blocks.0.mlp.fc1" in lora_model.loras
@@ -152,7 +163,8 @@ def test_legacy_manager_uses_the_h3_model_loader_without_changing_its_interface(
 
     class _Pipeline:
         def _load_diffusion_lora_adapter(self, **kwargs):
-            return load_minimax_h3_turbo_lora(partition="fl2va", **kwargs)
+            lora_model, peft_helper, _spec_unused = load_minimax_h3_turbo_lora(partition="fl2va", **kwargs)
+            return lora_model, peft_helper
 
     manager = object.__new__(DiffusionLoRAManager)
     manager.pipeline = _Pipeline()
@@ -168,22 +180,30 @@ def test_legacy_manager_uses_the_h3_model_loader_without_changing_its_interface(
     assert "blocks.0.mlp.fc1" in lora_model.loras
 
 
-def test_h3_turbo_rejects_wrong_alpha_and_ref2va(tmp_path):
-    wrong_alpha = tmp_path / "wrong_alpha" / "minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors"
-    wrong_alpha.parent.mkdir()
-    _write_tiny_turbo(wrong_alpha, alpha="8")
-    with pytest.raises(ValueError, match="requires alpha=128"):
-        load_minimax_h3_turbo_lora(
-            partition="fl2va",
-            lora_request=_request(wrong_alpha),
-            lora_path=wrong_alpha,
-            dtype=torch.float32,
-        )
+def test_h3_turbo_honours_the_declared_alpha_and_task_family(tmp_path):
+    """Alpha comes from the artifact; the family decides which server serves it."""
+
+    eight_step_alpha = tmp_path / "alpha8" / "minimax_h3_fl2v_turbo_8step_v1.0_768p_bf16.safetensors"
+    eight_step_alpha.parent.mkdir()
+    _write_tiny_turbo(eight_step_alpha, alpha="8")
+    loaded = load_minimax_h3_turbo_lora(
+        partition="fl2va",
+        lora_request=_request(eight_step_alpha),
+        lora_path=eight_step_alpha,
+        dtype=torch.float32,
+    )
+    assert loaded is not None
+    _, peft_helper, spec = loaded
+    # The eight-step artifact is a 1/16-strength adapter; pinning alpha to the
+    # four-step value would over-drive it by 16x.
+    assert peft_helper.lora_alpha == 8
+    assert spec.alpha == 8
+    assert spec.sigma_points == 9
 
     valid = tmp_path / "valid" / "minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors"
     valid.parent.mkdir()
     _write_tiny_turbo(valid)
-    with pytest.raises(ValueError, match="supports FL2VA/T2VA only"):
+    with pytest.raises(ValueError, match="Ref2VA-only server cannot serve it"):
         load_minimax_h3_turbo_lora(
             partition="ref2va",
             lora_request=_request(valid),
@@ -192,6 +212,43 @@ def test_h3_turbo_rejects_wrong_alpha_and_ref2va(tmp_path):
         )
 
 
+@pytest.mark.parametrize("partition", ["fl2va", "combined"])
+def test_h3_turbo_ref2v_artifact_needs_a_ref2va_only_server(tmp_path, partition):
+    """A combined server routes ref2va to ``transformers_ref``, which the
+    adapter cannot bind to: the LoRA target pattern only injects into
+    ``transformer``. Admitting it there would run an undistilled DiT on the
+    artifact's few-step schedule with no error anywhere."""
+
+    path = tmp_path / "minimax_h3_ref2v_turbo_8step_v1.0_768p_bf16.safetensors"
+    _write_tiny_turbo(path)
+
+    with pytest.raises(ValueError, match="start the server with --task-type ref2va"):
+        load_minimax_h3_turbo_lora(
+            partition=partition,
+            lora_request=_request(path),
+            lora_path=path,
+            dtype=torch.float32,
+        )
+
+
+def test_h3_turbo_ref2v_artifact_loads_on_a_ref2va_server(tmp_path):
+    path = tmp_path / "minimax_h3_ref2v_turbo_8step_v1.0_768p_bf16.safetensors"
+    _write_tiny_turbo(path)
+
+    loaded = load_minimax_h3_turbo_lora(
+        partition="ref2va",
+        lora_request=_request(path),
+        lora_path=path,
+        dtype=torch.float32,
+    )
+
+    assert loaded is not None
+    spec = loaded[2]
+    assert spec.task_family == "ref2v"
+    assert spec.supported_tasks == frozenset({"ref2va"})
+    assert spec.sigma_points == 9
+
+
 @pytest.mark.parametrize(
     "offload_mode",
     [
@@ -214,6 +271,8 @@ def test_h3_turbo_rejects_offload_modes(tmp_path, offload_mode):
 
 
 def test_h3_turbo_allows_distributed_layerwise_offload(monkeypatch):
+    """DLO keeps the LoRA buffers resident, so it is not an unsupported mode."""
+
     from vllm_omni.diffusion.models.minimax_h3 import MiniMaxH3Pipeline
     from vllm_omni.diffusion.models.minimax_h3 import (
         pipeline_minimax_h3 as pipeline_module,
@@ -222,7 +281,7 @@ def test_h3_turbo_allows_distributed_layerwise_offload(monkeypatch):
     pipeline = object.__new__(MiniMaxH3Pipeline)
     torch.nn.Module.__init__(pipeline)
     pipeline.partition = "fl2va"
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = set()
     pipeline._lora_sigma_schedules = {}
     pipeline.od_config = SimpleNamespace(
@@ -231,10 +290,11 @@ def test_h3_turbo_allows_distributed_layerwise_offload(monkeypatch):
         enable_distributed_layerwise_offload=True,
     )
     captured = {}
+    spec = _spec()
 
     def load_turbo(**kwargs):
         captured.update(kwargs)
-        return object(), object()
+        return object(), object(), spec
 
     monkeypatch.setattr(pipeline_module, "load_minimax_h3_turbo_lora", load_turbo)
     loaded = pipeline._load_diffusion_lora_adapter(
@@ -245,6 +305,10 @@ def test_h3_turbo_allows_distributed_layerwise_offload(monkeypatch):
 
     assert loaded is not None
     assert captured["unsupported_offload_mode"] is None
+    # The pipeline passes its own partition through and keeps the spec that
+    # decides the request contract.
+    assert captured["partition"] == "fl2va"
+    assert pipeline._turbo_lora_specs == {1: spec}
 
 
 @pytest.mark.parametrize(
@@ -268,7 +332,7 @@ def test_h3_turbo_offload_compatibility_follows_dit_selection(
     pipeline = object.__new__(MiniMaxH3Pipeline)
     torch.nn.Module.__init__(pipeline)
     pipeline.partition = "fl2va"
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = set()
     pipeline._lora_sigma_schedules = {}
     pipeline.od_config = SimpleNamespace(
@@ -281,10 +345,11 @@ def test_h3_turbo_offload_compatibility_follows_dit_selection(
         pin_cpu_memory=True,
     )
     captured = {}
+    spec = _spec()
 
     def load_turbo(**kwargs):
         captured.update(kwargs)
-        return object(), object()
+        return object(), object(), spec
 
     monkeypatch.setattr(pipeline_module, "load_minimax_h3_turbo_lora", load_turbo)
 
@@ -296,31 +361,84 @@ def test_h3_turbo_offload_compatibility_follows_dit_selection(
     assert captured["unsupported_offload_mode"] == expected_mode
 
 
-def test_h3_turbo_accepts_only_the_declared_v1_artifact(tmp_path):
-    unsupported = tmp_path / "minimax_h3_fl2v_turbo_8step_v1.0_bf16.safetensors"
-    _write_tiny_turbo(unsupported)
+def test_h3_turbo_uses_the_reference_alpha_when_the_artifact_declares_none(tmp_path):
+    """``4step_v0.1`` ships no alpha. LightX2V's reference script applies
+    ``scale * alpha / rank`` with alpha defaulting to 8, so the fallback must be
+    8; alpha == rank would drive the adapter 16x too strongly."""
 
-    with pytest.raises(ValueError, match="supports only"):
+    path = tmp_path / "minimax_h3_fl2v_turbo_4step_v0.1.safetensors"
+    _write_tiny_turbo(path, alpha=None)
+
+    loaded = load_minimax_h3_turbo_lora(
+        partition="fl2va",
+        lora_request=_request(path),
+        lora_path=path,
+        dtype=torch.float32,
+    )
+
+    assert loaded is not None
+    _, peft_helper, spec = loaded
+    assert spec.alpha == 8
+    assert peft_helper.lora_alpha == 8
+    assert spec.alpha / spec.rank == 0.0625
+
+
+def test_h3_turbo_declines_the_comfyui_export(tmp_path):
+    """The ComfyUI exports fuse Q/K/V, so they are not this loader's format."""
+
+    path = tmp_path / "minimax_h3_fl2v_turbo_4step_v1.0_768p_comfyui_bf16.safetensors"
+    _write_tiny_turbo(path)
+
+    assert (
         load_minimax_h3_turbo_lora(
             partition="fl2va",
-            lora_request=_request(unsupported),
-            lora_path=unsupported,
+            lora_request=_request(path),
+            lora_path=path,
             dtype=torch.float32,
         )
+        is None
+    )
 
-    supported = tmp_path / "minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors"
-    _write_tiny_turbo(supported)
-    # Directory resolution selects the declared artifact even when another
-    # v1.0 checkpoint is present beside it.
-    assert (
+
+def test_h3_turbo_accepts_every_published_artifact_name(tmp_path):
+    """Each published filename loads and carries its own sampler contract."""
+
+    for index, (name, steps, shift) in enumerate(
+        (
+            ("minimax_h3_fl2v_turbo_4step_v0.1.safetensors", 4, 12.0),
+            ("minimax_h3_fl2v_turbo_4step_v1.1_768p_bf16.safetensors", 4, 6.0),
+            ("minimax_h3_fl2v_turbo_8step_v1.0_bf16.safetensors", 8, 12.0),
+            ("minimax_h3_fl2v_turbo_8step_v1.0_768p_bf16.safetensors", 8, 6.0),
+        )
+    ):
+        directory = tmp_path / f"artifact{index}"
+        directory.mkdir()
+        path = directory / name
+        _write_tiny_turbo(path)
+        loaded = load_minimax_h3_turbo_lora(
+            partition="fl2va",
+            lora_request=_request(path),
+            lora_path=path,
+            dtype=torch.float32,
+        )
+        assert loaded is not None, name
+        spec = loaded[2]
+        assert spec.denoise_steps == steps
+        assert spec.sigma_points == steps + 1
+        assert spec.video_shift == shift
+
+
+def test_h3_turbo_directory_with_several_artifacts_is_ambiguous(tmp_path):
+    _write_tiny_turbo(tmp_path / "minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors")
+    _write_tiny_turbo(tmp_path / "minimax_h3_fl2v_turbo_8step_v1.0_768p_bf16.safetensors")
+
+    with pytest.raises(ValueError, match="point --lora-path at one file"):
         load_minimax_h3_turbo_lora(
             partition="fl2va",
             lora_request=_request(tmp_path),
             lora_path=tmp_path,
             dtype=torch.float32,
         )
-        is not None
-    )
 
 
 def test_h3_turbo_rejects_a_truncated_declared_artifact(tmp_path):
@@ -352,7 +470,7 @@ def test_non_h3_checkpoint_falls_back_to_the_generic_peft_loader(tmp_path):
 
     declared = tmp_path / "minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors"
     _write_tiny_turbo(declared, key_format="other")
-    with pytest.raises(ValueError, match="requires safetensors metadata"):
+    with pytest.raises(ValueError, match="expected 'minimax-h3-diffusers'"):
         load_minimax_h3_turbo_lora(
             partition="fl2va",
             lora_request=_request(declared),
@@ -371,7 +489,7 @@ def test_only_an_active_recognized_turbo_adapter_restricts_ref2va(monkeypatch):
     torch.nn.Module.__init__(pipeline)
     pipeline.partition = "combined"
     pipeline.supported_tasks = frozenset({"t2va", "fl2va", "ref2va"})
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = set()
     pipeline._lora_sigma_schedules = {}
     request = _request("generic-peft")
@@ -388,14 +506,14 @@ def test_only_an_active_recognized_turbo_adapter_restricts_ref2va(monkeypatch):
         return pipeline._resolve_task(
             "ref2va",
             {},
-            has_turbo_lora=pipeline._has_active_turbo_lora(sampling),
+            turbo_spec=pipeline._active_turbo_spec(sampling),
         )
 
-    recognized = (object(), object())
+    recognized = (object(), object(), _spec())
     monkeypatch.setattr(pipeline_module, "load_minimax_h3_turbo_lora", lambda **_: recognized)
-    assert load() is recognized
+    assert load() == recognized[:2]
     assert resolve(0.0) == "ref2va"
-    with pytest.raises(OmniClientError, match="supports T2VA/FL2VA requests only"):
+    with pytest.raises(OmniClientError, match="serves"):
         resolve(1.0)
 
     # Simulate manager eviction followed by a generic PEFT adapter reusing the
@@ -410,7 +528,7 @@ def test_h3_turbo_requires_all_loaded_targets_to_bind():
 
     pipeline = object.__new__(MiniMaxH3Pipeline)
     torch.nn.Module.__init__(pipeline)
-    pipeline._turbo_lora_adapter_ids = {1}
+    pipeline._turbo_lora_specs = {1: _spec()}
     lora_model = SimpleNamespace(id=1, loras={"to_q": object(), "to_k": object()})
 
     pipeline._validate_diffusion_lora_binding(
@@ -448,7 +566,7 @@ def test_h3_turbo_rejects_unsupported_sampling(num_inference_steps, extra_args,
     )
 
     with pytest.raises(OmniClientError, match=error):
-        pipeline._validate_turbo_sampling(sampling)
+        pipeline._validate_turbo_sampling(sampling, _spec())
 
 
 def test_h3_turbo_accepts_five_sigma_points_for_four_nfe():
@@ -461,5 +579,6 @@ def test_h3_turbo_accepts_five_sigma_points_for_four_nfe():
         SimpleNamespace(
             num_inference_steps=5,
             extra_args={"flow_shift": 6.0, "audio_flow_shift": 3.0},
-        )
+        ),
+        _spec(),
     )

tests/diffusion/models/minimax_h3/test_minimax_h3_native_lora.py

@@ -309,7 +309,7 @@ def test_pipeline_native_schedule_and_task_validation(monkeypatch):
     torch.nn.Module.__init__(pipeline)
     pipeline.partition = "fl2va"
     pipeline.supported_tasks = frozenset({"t2va", "fl2va", "ref2va"})
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = {7}
     pipeline._lora_sigma_schedules = {7: DMD2SigmaSchedule.from_positions([1.0, 0.7, 0.4, 0.15, 0.0])}
     pipeline._base_schedule_by_partition = {"fl2va": None}
@@ -330,7 +330,7 @@ def test_pipeline_native_schedule_and_task_validation(monkeypatch):
         pipeline._resolve_task(
             "fl2va",
             {},
-            has_turbo_lora=False,
+            turbo_spec=None,
             has_native_lora=True,
         )
 
@@ -418,7 +418,7 @@ def test_pipeline_replaces_native_classification_after_reload(monkeypatch, tmp_p
         enable_layerwise_offload=False,
         enable_distributed_layerwise_offload=False,
     )
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = set()
     pipeline._lora_sigma_schedules = {}
 
@@ -472,7 +472,7 @@ def test_h3_native_allows_distributed_layerwise_offload(monkeypatch):
         enable_layerwise_offload=False,
         enable_distributed_layerwise_offload=True,
     )
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = set()
     pipeline._lora_sigma_schedules = {}
     captured: dict[str, object] = {}
@@ -612,7 +612,7 @@ def test_legacy_manager_uses_native_loader(tmp_path):
                 enable_layerwise_offload=False,
                 enable_distributed_layerwise_offload=False,
             )
-            pipeline._turbo_lora_adapter_ids = set()
+            pipeline._turbo_lora_specs = {}
             pipeline._native_lora_adapter_ids = set()
             pipeline._lora_sigma_schedules = {}
             return pipeline._load_diffusion_lora_adapter(**kwargs)
@@ -634,7 +634,7 @@ def test_pipeline_schedule_inactive_when_scale_zero():
     pipeline = object.__new__(MiniMaxH3Pipeline)
     torch.nn.Module.__init__(pipeline)
     pipeline.partition = "fl2va"
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = {7}
     pipeline._lora_sigma_schedules = {7: DMD2SigmaSchedule.from_positions([1.0, 0.7, 0.4, 0.15, 0.0])}
     pipeline._base_schedule_by_partition = {"fl2va": None}
@@ -664,7 +664,7 @@ def test_pipeline_schedule_falls_back_after_eviction(monkeypatch, tmp_path):
         enable_layerwise_offload=False,
         enable_distributed_layerwise_offload=False,
     )
-    pipeline._turbo_lora_adapter_ids = set()
+    pipeline._turbo_lora_specs = {}
     pipeline._native_lora_adapter_ids = set()
     pipeline._lora_sigma_schedules = {}
     pipeline._base_schedule_by_partition = {"fl2va": None}

tests/diffusion/models/minimax_h3/test_minimax_h3_turbo_matrix.py

@@ -0,0 +1,86 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+
+"""Contract coverage for the LightX2V MiniMax-H3 Turbo artifact matrix."""
+
+from __future__ import annotations
+
+import pytest
+
+from vllm_omni.diffusion.models.minimax_h3.lora import (
+    _TURBO_AUDIO_SHIFT,
+    _TURBO_VIDEO_SHIFT_544P,
+    _TURBO_VIDEO_SHIFT_768P,
+    TurboSpec,
+    parse_turbo_filename,
+)
+
+pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu]
+
+# The eight Diffusers-layout artifacts published at lightx2v/Minimax-h3-Turbo,
+# with the contract each one implies. The ComfyUI export of an artifact is a
+# fused-QKV re-packing of the same weights and is not served.
+PUBLISHED = [
+    ("minimax_h3_fl2v_turbo_4step_v0.1.safetensors", "fl2v", 4, _TURBO_VIDEO_SHIFT_544P),
+    ("minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors", "fl2v", 4, _TURBO_VIDEO_SHIFT_768P),
+    ("minimax_h3_fl2v_turbo_4step_v1.1_768p_bf16.safetensors", "fl2v", 4, _TURBO_VIDEO_SHIFT_768P),
+    ("minimax_h3_fl2v_turbo_4step_v1.2_768p_bf16.safetensors", "fl2v", 4, _TURBO_VIDEO_SHIFT_768P),
+    ("minimax_h3_fl2v_turbo_8step_v1.0_bf16.safetensors", "fl2v", 8, _TURBO_VIDEO_SHIFT_544P),
+    ("minimax_h3_fl2v_turbo_8step_v1.0_768p_bf16.safetensors", "fl2v", 8, _TURBO_VIDEO_SHIFT_768P),
+    ("minimax_h3_ref2v_turbo_4step_v0.1_bf16.safetensors", "ref2v", 4, _TURBO_VIDEO_SHIFT_544P),
+    ("minimax_h3_ref2v_turbo_8step_v1.0_768p_bf16.safetensors", "ref2v", 8, _TURBO_VIDEO_SHIFT_768P),
+]
+
+
+@pytest.mark.parametrize(("filename", "task", "steps", "video_shift"), PUBLISHED)
+def test_every_published_artifact_is_recognised(filename: str, task: str, steps: int, video_shift: float) -> None:
+    spec = parse_turbo_filename(filename)
+    assert spec is not None, filename
+    assert spec.filename == filename
+    assert spec.task_family == task
+    assert spec.denoise_steps == steps
+    assert spec.video_shift == video_shift
+    assert spec.audio_shift == _TURBO_AUDIO_SHIFT
+    # Until the loader reads metadata, alpha is LightX2V's reference default.
+    assert spec.rank == 128
+    assert spec.alpha == 8.0
+
+
+def test_all_eight_artifacts_are_covered() -> None:
+    assert len(PUBLISHED) == 8
+    assert len({name for name, *_ in PUBLISHED}) == 8
+
+
+@pytest.mark.parametrize(
+    "filename",
+    [
+        "adapter_model.safetensors",
+        "minimax_h3_fl2v_turbo_v1.0_768p_bf16.safetensors",
+        "minimax_h3_t2v_turbo_4step_v1.0.safetensors",
+        "minimax_h3_fl2v_turbo_0step_v1.0.safetensors",
+        # The ComfyUI exports are a different tensor layout, not a variant of
+        # the supported one; the loader refuses them by name.
+        "minimax_h3_fl2v_turbo_4step_v1.0_768p_comfyui_bf16.safetensors",
+        "minimax_h3_ref2v_turbo_8step_v1.0_768p_comfyui_bf16.safetensors",
+    ],
+)
+def test_unrelated_names_are_rejected(filename: str) -> None:
+    assert parse_turbo_filename(filename) is None
+
+
+def _spec(filename: str) -> TurboSpec:
+    spec = parse_turbo_filename(filename)
+    assert spec is not None
+    return spec
+
+
+def test_sigma_points_follow_the_step_count() -> None:
+    assert _spec("minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors").sigma_points == 5
+    assert _spec("minimax_h3_fl2v_turbo_8step_v1.0_768p_bf16.safetensors").sigma_points == 9
+
+
+def test_task_family_decides_supported_tasks() -> None:
+    fl2v = _spec("minimax_h3_fl2v_turbo_4step_v1.0_768p_bf16.safetensors")
+    ref2v = _spec("minimax_h3_ref2v_turbo_8step_v1.0_768p_bf16.safetensors")
+    assert fl2v.supported_tasks == frozenset({"t2va", "fl2va"})
+    assert ref2v.supported_tasks == frozenset({"ref2va"})