fix(wan): pair Wan 2.2 A14B GGUF experts by wiring, not just filename

invokeai/app/invocations/wan_model_loader.py

@@ -154,6 +154,11 @@ def invoke(self, context: InvocationContext) -> WanModelLoaderOutput:
                 context.logger.warning("'Transformer (Low Noise)' is ignored for the single-expert TI2V-5B variant.")
 
             if self.transformer_low_noise_model is not None and main_variant != WanVariantType.TI2V_5B:
+                if self.transformer_low_noise_model.key == self.model.key:
+                    raise ValueError(
+                        "The same model is wired to both 'Transformer' and 'Transformer (Low Noise)'. "
+                        "A Wan A14B expert pair needs two different GGUF models."
+                    )
                 low_config = context.models.get_config(self.transformer_low_noise_model)
                 self._validate_main_config(low_config, "Transformer (Low Noise)")
                 if low_config.format != ModelFormat.GGUFQuantized:
@@ -166,30 +171,63 @@ def invoke(self, context: InvocationContext) -> WanModelLoaderOutput:
 
                 if getattr(low_config, "variant", None) != main_variant:
                     raise ValueError("The high-noise and low-noise GGUF models must use the same Wan variant.")
-                if {primary_expert, low_expert} != {"high", "low"}:
-                    raise ValueError("A Wan A14B GGUF expert pair must contain one high and one low expert.")
+
+                # The expert tag is a filename heuristic, so 'none' (untagged)
+                # is common on community finetunes. The wiring itself is
+                # explicit user intent — main slot = high, low-noise slot =
+                # low — so an untagged file is taken at its wired position (or
+                # inferred as the complement of its tagged partner). Only a
+                # genuine conflict, both files claiming the *same* expert, is
+                # an error.
+                if primary_expert == low_expert != "none":
+                    raise ValueError(
+                        f"Both selected GGUF models are tagged as the {primary_expert}-noise expert. "
+                        "A Wan A14B expert pair must contain one high and one low expert."
+                    )
+                if primary_expert == "none" and low_expert == "none":
+                    context.logger.warning(
+                        "Neither Wan A14B GGUF filename identifies its expert, so 'Transformer' is assumed to "
+                        "be the high-noise expert and 'Transformer (Low Noise)' the low-noise expert. If the "
+                        "output looks wrong, swap the two models."
+                    )
 
                 # Make sure 'transformer' is the high-noise expert and
                 # 'transformer_low_noise' is the low-noise expert. If the user
                 # accidentally swapped them, swap back.
-                if primary_expert == "low" and low_expert == "high":
+                if primary_expert == "low" or low_expert == "high":
                     transformer = low_id
                     transformer_low_noise = primary_id
+                    # The swap overrides the wiring on the strength of a
+                    # filename tag, so say so: a mistagged file is otherwise an
+                    # invisible expert inversion.
+                    context.logger.warning(
+                        f"The wired Wan A14B GGUF experts look reversed, so they were swapped: "
+                        f"'{low_config.name}' (tagged '{low_expert}') runs as the high-noise expert and "
+                        f"'{main_config.name}' (tagged '{primary_expert}') as the low-noise expert. "
+                        "The tags come from the filenames — if the output looks wrong, a filename is lying."
+                    )
                 else:
                     transformer = primary_id
                     transformer_low_noise = low_id
             else:
                 transformer = primary_id
-                if main_variant in (WanVariantType.T2V_A14B, WanVariantType.I2V_A14B) and primary_expert != "high":
-                    raise ValueError("An unpaired Wan A14B GGUF model must be the high-noise expert.")
                 # A14B without a paired low-noise GGUF will produce degraded
-                # quality (only the high-noise expert runs). Warn but don't
-                # abort — TI2V-5B GGUFs are single-expert and totally fine.
+                # quality (only one expert runs). Warn but don't abort — a
+                # single wired transformer is explicit intent just like a pair
+                # is, and the tag is only a filename guess, so an untagged file
+                # must not be fatal here when the paired path accepts it.
+                # TI2V-5B GGUFs are single-expert and totally fine.
                 if main_variant in (WanVariantType.T2V_A14B, WanVariantType.I2V_A14B):
-                    context.logger.warning(
-                        "A14B GGUF main was provided without a paired 'Transformer (Low Noise)'. "
-                        "Only the high-noise expert will run; image quality will be reduced."
+                    message = (
+                        "An A14B GGUF is wired to 'Transformer' without a paired 'Transformer (Low Noise)'. "
+                        "Only this one expert will run; image quality will be reduced."
                     )
+                    if primary_expert == "low":
+                        message += (
+                            " Its filename tags it as the low-noise expert; when running a single expert, "
+                            "the high-noise one is usually the better choice."
+                        )
+                    context.logger.warning(message)
 
             # Borrow the boundary_ratio recorded on the optional Diffusers
             # component_source, when one is wired.

tests/app/invocations/test_wan_model_loader.py

@@ -35,7 +35,7 @@ def _config(
     )
 
 
-def _invoke(
+def _prepare(
     main_config: SimpleNamespace,
     low_config: SimpleNamespace | None = None,
     component_config: SimpleNamespace | None = None,
@@ -44,13 +44,14 @@ def _invoke(
     vae_latent_channels: int | None = None,
     vae_config: SimpleNamespace | None = None,
     t5_config: SimpleNamespace | None = None,
-):
+    low_key: str = "low",
+) -> tuple[WanModelLoaderInvocation, MagicMock]:
     main = _model("main")
-    low = _model("low") if low_config is not None else None
+    low = _model(low_key) if low_config is not None else None
     context = MagicMock()
     configs = {"main": main_config}
     if low_config is not None:
-        configs["low"] = low_config
+        configs[low_key] = low_config
     component = _model("component") if component_config is not None else None
     if component_config is not None:
         configs["component"] = component_config
@@ -78,9 +79,18 @@ def _invoke(
         wan_t5_encoder_model=_model("t5"),
         component_source=component,
     )
+    return invocation, context
+
+
+def _invoke(*args, **kwargs):
+    invocation, context = _prepare(*args, **kwargs)
     return invocation.invoke(context)
 
 
+def _warnings(context: MagicMock) -> list[str]:
+    return [call.args[0] for call in context.logger.warning.call_args_list]
+
+
 @pytest.mark.parametrize("variant", [WanVariantType.T2V_A14B, WanVariantType.I2V_A14B])
 @pytest.mark.parametrize("main_expert,low_expert", [("high", "low"), ("low", "high")])
 def test_gguf_loader_accepts_valid_expert_pair_in_either_order(
@@ -108,8 +118,8 @@ def test_gguf_loader_accepts_valid_expert_pair_in_either_order(
             _config("low", WanVariantType.T2V_A14B, "high"),
         ),
         (
-            _config("main", WanVariantType.T2V_A14B, "high"),
-            _config("low", WanVariantType.T2V_A14B, "none"),
+            _config("main", WanVariantType.T2V_A14B, "low"),
+            _config("low", WanVariantType.T2V_A14B, "low"),
         ),
     ],
 )
@@ -118,6 +128,33 @@ def test_gguf_loader_rejects_invalid_expert_pair(main_config: SimpleNamespace, l
         _invoke(main_config, low_config)
 
 
+@pytest.mark.parametrize(
+    "main_expert,low_expert,expected_high_key",
+    [
+        # The expert tag comes from a filename heuristic, so untagged community
+        # finetunes are common. The wiring is explicit intent: take the untagged
+        # file at its wired position, or as the complement of a tagged partner.
+        ("none", "none", "main"),
+        ("high", "none", "main"),
+        ("none", "low", "main"),
+        ("none", "high", "low"),
+        ("low", "none", "low"),
+    ],
+)
+def test_gguf_loader_falls_back_to_wiring_for_untagged_experts(
+    main_expert: str, low_expert: str, expected_high_key: str
+) -> None:
+    output = _invoke(
+        _config("main", WanVariantType.I2V_A14B, main_expert),
+        _config("low", WanVariantType.I2V_A14B, low_expert),
+    )
+
+    expected_low_key = "low" if expected_high_key == "main" else "main"
+    assert output.transformer.transformer.key == expected_high_key
+    assert output.transformer.transformer_low_noise is not None
+    assert output.transformer.transformer_low_noise.key == expected_low_key
+
+
 @pytest.mark.parametrize("low_variant", [WanVariantType.TI2V_5B, WanVariantType.T2V_A14B])
 def test_ti2v_5b_main_ignores_wired_low_noise_model(low_variant: WanVariantType) -> None:
     """The field docs promise 'Transformer (Low Noise)' is ignored for the single-expert
@@ -131,10 +168,66 @@ def test_ti2v_5b_main_ignores_wired_low_noise_model(low_variant: WanVariantType)
     assert output.transformer.transformer_low_noise is None
 
 
-@pytest.mark.parametrize("expert", ["low", "none"])
-def test_gguf_loader_rejects_non_high_primary_without_pair(expert: str) -> None:
-    with pytest.raises(ValueError, match="high-noise"):
-        _invoke(_config("main", WanVariantType.T2V_A14B, expert))
+@pytest.mark.parametrize("expert", ["high", "low", "none"])
+def test_gguf_loader_runs_unpaired_primary_whatever_its_tag(expert: str) -> None:
+    """A single wired transformer is explicit intent just like a pair is, and the tag is
+    only a filename guess — so an unpaired A14B runs with a warning rather than aborting."""
+    invocation, context = _prepare(_config("main", WanVariantType.T2V_A14B, expert))
+    output = invocation.invoke(context)
+
+    assert output.transformer.transformer.key == "main"
+    assert output.transformer.transformer_low_noise is None
+    assert any("only this one expert will run" in warning.lower() for warning in _warnings(context))
+
+
+def test_gguf_loader_hints_at_the_expert_swap_for_an_unpaired_low_noise_model() -> None:
+    invocation, context = _prepare(_config("main", WanVariantType.T2V_A14B, "low"))
+    invocation.invoke(context)
+
+    assert any("high-noise one is usually the better choice" in warning for warning in _warnings(context))
+
+
+def test_gguf_loader_rejects_the_same_model_in_both_transformer_slots() -> None:
+    """Wiring one model twice used to fail the {high, low} pair check. It must stay an error:
+    the denoiser would unload and reload the same multi-GB expert at every boundary crossing."""
+    main_config = _config("main", WanVariantType.T2V_A14B, "high")
+    with pytest.raises(ValueError, match="same model"):
+        _invoke(main_config, main_config, low_key="main")
+
+
+@pytest.mark.parametrize("main_expert,low_expert", [("low", "high"), ("low", "none"), ("none", "high")])
+def test_gguf_loader_warns_when_it_swaps_the_wired_experts(main_expert: str, low_expert: str) -> None:
+    """The swap overrides explicit wiring on the strength of a filename tag, so a mistagged
+    file must not invert the two experts silently."""
+    invocation, context = _prepare(
+        _config("main", WanVariantType.I2V_A14B, main_expert),
+        _config("low", WanVariantType.I2V_A14B, low_expert),
+    )
+    output = invocation.invoke(context)
+
+    assert output.transformer.transformer.key == "low"
+    assert any("swapped" in warning for warning in _warnings(context))
+
+
+@pytest.mark.parametrize("main_expert,low_expert", [("high", "low"), ("high", "none"), ("none", "low")])
+def test_gguf_loader_is_quiet_when_the_wiring_stands(main_expert: str, low_expert: str) -> None:
+    invocation, context = _prepare(
+        _config("main", WanVariantType.I2V_A14B, main_expert),
+        _config("low", WanVariantType.I2V_A14B, low_expert),
+    )
+    invocation.invoke(context)
+
+    assert _warnings(context) == []
+
+
+def test_gguf_loader_warns_when_neither_expert_is_tagged() -> None:
+    invocation, context = _prepare(
+        _config("main", WanVariantType.I2V_A14B, "none"),
+        _config("low", WanVariantType.I2V_A14B, "none"),
+    )
+    invocation.invoke(context)
+
+    assert any("Neither Wan A14B GGUF filename identifies its expert" in warning for warning in _warnings(context))
 
 
 @pytest.mark.parametrize(