[Perf][Diffusion] Fuse lossless Wan VAE data movement

Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>

tests/diffusion/distributed/test_autoencoder_kl_wan.py

@@ -1,3 +1,5 @@
+from types import SimpleNamespace
+
 import pytest
 import torch
 
@@ -41,3 +43,21 @@ def test_wan_vae_execution_context_uses_platform_autocast(mocker):
         dtype=torch.bfloat16,
         enabled=True,
     )
+
+
+def test_prepare_pipeline_installs_data_movement_before_compile(monkeypatch):
+    model = _DummyOmniAutoencoderKLWan(dtype=torch.float32)
+    installed = []
+    monkeypatch.setattr(
+        "vllm_omni.diffusion.offloader.module_collector.ModuleDiscovery.discover",
+        lambda pipeline: SimpleNamespace(vaes=[model]),
+    )
+    monkeypatch.setattr(
+        wan_vae_module,
+        "install_wan_vae_data_movement",
+        lambda vae: installed.append(vae),
+    )
+
+    wan_vae_module.prepare_pipeline_wan_vae_data_movement(torch.nn.Identity())
+
+    assert installed == [model]

tests/diffusion/kernels/test_wan_vae_data_movement.py

@@ -0,0 +1,274 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+import pytest
+import torch
+import torch.nn.functional as F
+from diffusers.models.autoencoders import AutoencoderKLWan
+
+from vllm_omni.diffusion.distributed.autoencoders import wan_vae_data_movement
+from vllm_omni.diffusion.distributed.autoencoders.autoencoder_kl_wan import OmniAutoencoderKLWan
+from vllm_omni.diffusion.distributed.autoencoders.wan_vae_data_movement import (
+    _cache_payload,
+    _causal_padding,
+    _is_dynamic_spatial_conv,
+    _run_cached_causal_conv,
+    install_wan_vae_data_movement,
+)
+from vllm_omni.diffusion.kernels.wan_vae_data_movement import cat_pad_5d, dup_up3d_add
+
+pytestmark = pytest.mark.core_model
+cuda_only = pytest.mark.skipif(not torch.cuda.is_available(), reason="CUDA required")
+
+
+def _dense_5d(shape: tuple[int, ...], dtype: torch.dtype, channels_last: bool) -> torch.Tensor:
+    tensor = torch.randn(shape, device="cuda", dtype=dtype)
+    if channels_last:
+        return tensor.contiguous(memory_format=torch.channels_last_3d)
+    return tensor.contiguous()
+
+
+def _reference_cat_pad(
+    x: torch.Tensor,
+    cache: torch.Tensor | None,
+    padding: tuple[int, ...],
+) -> torch.Tensor:
+    reference_padding = list(padding)
+    if cache is not None:
+        x = torch.cat([cache, x], dim=2)
+        reference_padding[4] -= cache.shape[2]
+    return F.pad(x, reference_padding)
+
+
+@pytest.mark.cuda
+@cuda_only
+@torch.no_grad()
+@pytest.mark.parametrize("dtype", [torch.bfloat16, torch.float32])
+@pytest.mark.parametrize("channels_last", [False, True])
+@pytest.mark.parametrize(
+    "channels,frames,height,width,cache_frames,padding",
+    [
+        (96, 1, 10, 14, 0, (1, 1, 1, 1, 2, 0)),
+        (96, 1, 10, 14, 1, (1, 1, 1, 1, 2, 0)),
+        (96, 1, 10, 14, 2, (1, 1, 1, 1, 2, 0)),
+        (64, 1, 10, 14, 2, (0, 0, 0, 0, 2, 0)),
+        (48, 4, 10, 14, 2, (1, 1, 1, 1, 2, 0)),
+    ],
+)
+def test_cat_pad_5d_is_bitwise_and_layout_exact(
+    dtype: torch.dtype,
+    channels_last: bool,
+    channels: int,
+    frames: int,
+    height: int,
+    width: int,
+    cache_frames: int,
+    padding: tuple[int, ...],
+) -> None:
+    torch.cuda.manual_seed(0)
+    x = _dense_5d((1, channels, frames, height, width), dtype, channels_last)
+    cache = None
+    if cache_frames:
+        pad_height, pad_width = padding[2], padding[0]
+        buffer = _dense_5d(
+            (1, channels, cache_frames, height + 2 * pad_height, width + 2 * pad_width),
+            dtype,
+            channels_last,
+        )
+        cache = buffer[:, :, :, pad_height : pad_height + height, pad_width : pad_width + width]
+
+    reference = _reference_cat_pad(x, cache, padding)
+    output = cat_pad_5d(x, cache, padding)
+    assert output is not None
+    assert output.shape == reference.shape
+    assert output.stride() == reference.stride()
+    assert torch.equal(output, reference)
+
+    pair = cat_pad_5d(x, cache, padding, keep_cache_frames=2)
+    assert pair is not None
+    output_with_cache, next_cache = pair
+    assert torch.equal(output_with_cache, reference)
+    pad_height, pad_width = padding[2], padding[0]
+    keep_frames = min(2, reference.shape[2])
+    expected_cache = reference[
+        :,
+        :,
+        -keep_frames:,
+        pad_height : pad_height + height,
+        pad_width : pad_width + width,
+    ]
+    assert next_cache.shape == expected_cache.shape
+    assert torch.equal(next_cache, expected_cache)
+    assert next_cache.is_contiguous(memory_format=torch.channels_last_3d) is channels_last
+
+
+@pytest.mark.cuda
+@cuda_only
+@torch.no_grad()
+@pytest.mark.parametrize("dtype", [torch.bfloat16, torch.float32])
+@pytest.mark.parametrize("main_layout", ["contiguous", "channels_last", "permuted"])
+@pytest.mark.parametrize(
+    "in_channels,out_channels,frames,height,width,factor_temporal,factor_spatial,drop_first",
+    [
+        (128, 64, 1, 10, 14, 2, 2, False),
+        (128, 64, 1, 10, 14, 2, 2, True),
+        (64, 32, 2, 10, 14, 1, 2, False),
+    ],
+)
+def test_dup_up3d_add_is_bitwise_and_layout_exact(
+    dtype: torch.dtype,
+    main_layout: str,
+    in_channels: int,
+    out_channels: int,
+    frames: int,
+    height: int,
+    width: int,
+    factor_temporal: int,
+    factor_spatial: int,
+    drop_first: bool,
+) -> None:
+    torch.cuda.manual_seed(0)
+    repeats = out_channels * factor_temporal * factor_spatial * factor_spatial // in_channels
+    source = _dense_5d((1, in_channels, frames, height, width), dtype, main_layout == "channels_last")
+    out_frames = frames * factor_temporal - (factor_temporal - 1 if drop_first else 0)
+    shape = (1, out_channels, out_frames, height * factor_spatial, width * factor_spatial)
+    if main_layout == "channels_last":
+        main = _dense_5d(shape, dtype, True)
+    elif main_layout == "permuted":
+        main = torch.randn(
+            (shape[0], shape[2], shape[1], shape[3], shape[4]),
+            device="cuda",
+            dtype=dtype,
+        ).permute(0, 2, 1, 3, 4)
+    else:
+        main = _dense_5d(shape, dtype, False)
+
+    shortcut = source.repeat_interleave(repeats, dim=1)
+    shortcut = shortcut.view(
+        1,
+        out_channels,
+        factor_temporal,
+        factor_spatial,
+        factor_spatial,
+        frames,
+        height,
+        width,
+    )
+    shortcut = shortcut.permute(0, 1, 5, 2, 6, 3, 7, 4).contiguous()
+    shortcut = shortcut.view(shape[0], shape[1], frames * factor_temporal, shape[3], shape[4])
+    if drop_first:
+        shortcut = shortcut[:, :, factor_temporal - 1 :]
+    reference = main + shortcut
+
+    output = dup_up3d_add(
+        main,
+        source,
+        factor_temporal,
+        factor_spatial,
+        repeats,
+        drop_first,
+    )
+    assert output is not None
+    assert output.shape == reference.shape
+    assert output.stride() == reference.stride()
+    assert torch.equal(output, reference)
+
+
+@pytest.mark.cuda
+@cuda_only
+@torch.no_grad()
+@pytest.mark.parametrize("channels_last", [False, True])
+@pytest.mark.parametrize("temporal_only", [False, True])
+def test_cached_conv_chunk_loop_matches_reference(monkeypatch, channels_last: bool, temporal_only: bool) -> None:
+    from diffusers.models.autoencoders.autoencoder_kl_wan import CACHE_T, WanCausalConv3d
+
+    torch.cuda.manual_seed(0)
+    channels = 64
+    if temporal_only:
+        conv = WanCausalConv3d(channels, 2 * channels, (3, 1, 1), padding=(1, 0, 0))
+    else:
+        conv = WanCausalConv3d(channels, channels, 3, padding=1)
+    conv = conv.to(device="cuda", dtype=torch.float32)
+    chunks = [_dense_5d((1, channels, 1, 10, 14), torch.float32, channels_last) for _ in range(4)]
+
+    def run(force_reference: bool, start):
+        cache = [start]
+        outputs = []
+        original = wan_vae_data_movement.cat_pad_5d
+        if force_reference:
+            monkeypatch.setattr(wan_vae_data_movement, "cat_pad_5d", None)
+        try:
+            for chunk in chunks:
+                outputs.append(_run_cached_causal_conv(conv, chunk, cache, 0))
+        finally:
+            monkeypatch.setattr(wan_vae_data_movement, "cat_pad_5d", original)
+        return outputs, cache[0]
+
+    for start in (None, "Rep"):
+        fused_outputs, fused_cache = run(False, start)
+        reference_outputs, reference_cache = run(True, start)
+        for output, reference in zip(fused_outputs, reference_outputs, strict=True):
+            assert torch.equal(output, reference)
+        payload = _cache_payload(fused_cache)
+        assert payload is not None and payload.shape[2] == CACHE_T
+        assert torch.equal(payload, reference_cache[:, :, -CACHE_T:])
+
+
+@pytest.mark.cpu
+def test_installer_is_idempotent_and_preserves_state_dict_keys() -> None:
+    vae = OmniAutoencoderKLWan(
+        base_dim=8,
+        decoder_base_dim=8,
+        z_dim=4,
+        dim_mult=[1, 1],
+        num_res_blocks=1,
+        temperal_downsample=[False, False],
+        is_residual=True,
+    )
+    keys_before = set(vae.state_dict())
+    assert install_wan_vae_data_movement(vae)
+    assert install_wan_vae_data_movement(vae)
+    assert set(vae.state_dict()) == keys_before
+
+
+@pytest.mark.cpu
+def test_dynamic_spatial_conv_exposes_direct_path_compatibility_contract() -> None:
+    conv = torch.nn.Conv3d(2, 2, 3)
+    conv._vllm_omni_dynamic_spatial_shard_conv = True
+    conv._source_padding = (1, 1, 1, 1, 2, 0)
+
+    assert _is_dynamic_spatial_conv(conv)
+    assert _causal_padding(conv) == conv._source_padding
+
+
+@pytest.mark.cuda
+@cuda_only
+@torch.no_grad()
+@pytest.mark.parametrize("channels_last", [False, True])
+def test_tiny_wan_decoder_is_bitwise_exact(channels_last: bool) -> None:
+    config = dict(
+        base_dim=8,
+        decoder_base_dim=8,
+        z_dim=4,
+        dim_mult=[1, 1],
+        num_res_blocks=1,
+        temperal_downsample=[False, True],
+        is_residual=True,
+    )
+    torch.manual_seed(0)
+    reference = AutoencoderKLWan(**config).eval().to("cuda")
+    candidate = OmniAutoencoderKLWan(**config).eval().to("cuda")
+    candidate.load_state_dict(reference.state_dict())
+    if channels_last:
+        for model in (reference, candidate):
+            for module in model.decoder.modules():
+                if isinstance(module, torch.nn.Conv3d):
+                    module.weight.data = module.weight.data.contiguous(memory_format=torch.channels_last_3d)
+
+    latents = torch.randn(1, 4, 3, 4, 4, device="cuda")
+    if channels_last:
+        latents = latents.contiguous(memory_format=torch.channels_last_3d)
+    expected = reference.decode(latents, return_dict=False)[0]
+    output = candidate.decode(latents, return_dict=False)[0]
+    assert output.stride() == expected.stride()
+    assert torch.equal(output, expected)

tests/diffusion/test_diffusion_config_propagation.py

@@ -113,14 +113,6 @@ class TestCreateDefaultDiffusion:
         stages = StageConfigFactory.create_default_diffusion({"model": "x"})
         assert stages[0]["engine_args"]["cache_backend"] == "none"
 
-    def test_vae_decode_precision_roundtrip(self):
-        od = _roundtrip_diffusion_config(model="x", vae_decode_precision="bf16")
-        assert od.vae_decode_precision == "bf16"
-
-    def test_invalid_vae_decode_precision_is_rejected(self):
-        with pytest.raises(ValueError, match="vae_decode_precision"):
-            OmniDiffusionConfig(model="x", vae_decode_precision="int8")
-
     def test_single_gpu_default_devices(self):
         stages = StageConfigFactory.create_default_diffusion({"model": "x"})
         assert stages[0]["runtime"]["devices"] == "0"

tests/diffusion/test_diffusion_model_runner.py

@@ -512,35 +512,6 @@ def test_execute_model_accepts_bare_diffusion_output_from_single_request_pipelin
     assert runner.pipeline.last_req.num_reqs == 1
 
 
-@pytest.mark.core_model
-@pytest.mark.cpu
-@pytest.mark.parametrize(
-    ("quality", "expected_enabled"),
-    [(None, False), ("lossless", False), ("high", True)],
-)
-def test_execute_model_scopes_vae_fast_path_to_request_quality(monkeypatch, quality, expected_enabled):
-    runner = _make_runner(cache_backend=None, cache_backend_name="none")
-    req = _make_request()
-    req.sampling_params.quality = quality
-    gate_calls = []
-
-    @contextmanager
-    def _record_gate(pipeline, enabled):
-        gate_calls.append((pipeline, enabled, "enter"))
-        yield
-        gate_calls.append((pipeline, enabled, "exit"))
-
-    monkeypatch.setattr(model_runner_module, "set_forward_context", _noop_forward_context)
-    monkeypatch.setattr(model_runner_module, "use_pipeline_vae_fast_path", _record_gate)
-
-    DiffusionModelRunner.execute_model(runner, req)
-
-    assert gate_calls == [
-        (runner.pipeline, expected_enabled, "enter"),
-        (runner.pipeline, expected_enabled, "exit"),
-    ]
-
-
 class _BatchPipeline:
     """Pipeline returning a configurable list of outputs from forward()."""
 
@@ -608,18 +579,6 @@ def test_execute_model_batch_rejects_output_count_mismatch(monkeypatch):
         DiffusionModelRunner.execute_model_batch(runner, sched, runner.od_config)
 
 
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_execute_model_batch_rejects_mixed_vae_quality() -> None:
-    runner = _make_batch_runner(_BatchPipeline(outputs=[DiffusionOutput(output="a"), DiffusionOutput(output="b")]))
-    sched = _make_scheduler_output(num_reqs=2)
-    sched.scheduled_new_reqs[0].req.sampling_params.quality = "lossless"
-    sched.scheduled_new_reqs[1].req.sampling_params.quality = "high"
-
-    with pytest.raises(RuntimeError, match="cannot mix request quality levels"):
-        DiffusionModelRunner.execute_model_batch(runner, sched, runner.od_config)
-
-
 @pytest.mark.core_model
 @pytest.mark.cpu
 def test_execute_model_batch_routes_one_output_per_request(monkeypatch):
@@ -769,18 +728,7 @@ def test_execute_model_runs_forward_after_kv_receive(monkeypatch):
 @pytest.mark.core_model
 @pytest.mark.cpu
 def test_load_model_clears_cache_backend_for_unsupported_pipeline(monkeypatch):
-    memory_pool_active = False
+    prepared_pipelines = []
-    optimization_pool_states = []
-
-    @contextmanager
-    def _memory_pool_context(tag):
-        nonlocal memory_pool_active
-        assert tag == "weights"
-        memory_pool_active = True
-        try:
-            yield
-        finally:
-            memory_pool_active = False
 
     class _DummyLoader:
         def __init__(self, load_config, od_config=None):
@@ -831,20 +779,20 @@ def test_load_model_clears_cache_backend_for_unsupported_pipeline(monkeypatch):
     monkeypatch.setattr(model_runner_module, "DeviceMemoryProfiler", _DummyMemoryProfiler)
     monkeypatch.setattr(
         model_runner_module,
-        "optimize_pipeline_vaes",
+        "prepare_pipeline_wan_vae_data_movement",
-        lambda pipeline, od_config: optimization_pool_states.append(memory_pool_active),
+        lambda pipeline: prepared_pipelines.append(pipeline),
     )
     monkeypatch.setattr(model_runner_module, "get_offload_backend", lambda od_config, device: None)
     monkeypatch.setattr(
         model_runner_module, "get_cache_backend", lambda cache_backend, cache_config: dummy_cache_backend
     )
 
-    DiffusionModelRunner.load_model(runner, memory_pool_context_fn=_memory_pool_context)
+    DiffusionModelRunner.load_model(runner)
 
     assert runner.cache_backend is None
     assert runner.od_config.cache_backend is None
     assert dummy_cache_backend.enabled is False
-    assert optimization_pool_states == [True]
+    assert prepared_pipelines == [runner.pipeline]
 
 
 @pytest.mark.core_model

tests/diffusion/test_diffusion_worker.py

@@ -102,7 +102,6 @@ class TestDiffusionWorkerSleep:
 
     def test_sleep_level_1(self, mocker: MockerFixture, mock_gpu_worker):
         """Test sleep mode level 1 (offload weights only)."""
-        clear_vae_caches = mocker.patch("vllm_omni.diffusion.vae_optimizations.clear_pipeline_vae_fast_path_caches")
         mock_allocator_class = patch_cumem_allocator(mocker)
         mock_platform = mocker.patch("vllm_omni.diffusion.worker.diffusion_worker.current_omni_platform")
         mock_platform.get_free_memory.side_effect = [10 * 1024**3, 12 * 1024**3]
@@ -129,7 +128,6 @@ class TestDiffusionWorkerSleep:
 
         # Verify sleep was called with correct tags
         mock_allocator.sleep.assert_called_once_with(offload_tags=("weights",))
-        clear_vae_caches.assert_called_once_with(mock_gpu_worker.model_runner.pipeline)
         assert bool(result) is True
         # Verify buffers were NOT saved (level 1 doesn't save buffers)
         assert len(mock_gpu_worker._sleep_saved_buffers) == 0

tests/diffusion/vae_optimizations/test_vae_fastpaths.py

@@ -1,300 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Correctness contracts for request-gated VAE decoder fast paths."""
-
-from __future__ import annotations
-
-import pytest
-import torch
-import torch.nn.functional as F
-from torch import nn
-
-from tests.helpers.mark import hardware_test
-from vllm_omni.diffusion.vae_optimizations.gate import (
-    VaeFastPathGate,
-    register_vae_fast_path_gate,
-    use_pipeline_vae_fast_path,
-    use_vae_fast_path,
-)
-
-pytestmark = [pytest.mark.diffusion]
-
-
-class _PipelineWithVae(nn.Module):
-    def __init__(self, vae: nn.Module) -> None:
-        super().__init__()
-        self.vae = vae
-
-
-class _DistributedExecutorConfig:
-    def __init__(self, parallel_mode: str) -> None:
-        self.parallel_mode = parallel_mode
-
-
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_vae_fast_path_gate_restores_nested_state() -> None:
-    vae = nn.Identity()
-    gate = VaeFastPathGate()
-    register_vae_fast_path_gate(vae, gate)
-
-    with use_vae_fast_path(vae, True):
-        assert gate.enabled
-        with use_vae_fast_path(vae, False):
-            assert not gate.enabled
-        assert gate.enabled
-
-    assert not gate.enabled
-
-
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_pipeline_vae_fast_path_controls_discovered_vae() -> None:
-    vae = nn.Identity()
-    pipeline = _PipelineWithVae(vae)
-    gate = VaeFastPathGate()
-    register_vae_fast_path_gate(vae, gate)
-
-    with use_pipeline_vae_fast_path(pipeline, True):
-        assert gate.enabled
-
-    assert not gate.enabled
-
-
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_image_group_norm_fast_path_uses_compile_safe_fallback(monkeypatch) -> None:
-    from vllm_omni.diffusion.vae_optimizations import image
-
-    gate = VaeFastPathGate()
-    gate.enabled = True
-    norm = image.FusedGroupNormSiLU(nn.GroupNorm(2, 4), gate)
-    inputs = torch.randn(1, 4, 3, 3)
-
-    monkeypatch.setattr(torch.compiler, "is_compiling", lambda: True)
-    monkeypatch.setattr(
-        image,
-        "group_norm_silu_4d",
-        lambda *args, **kwargs: pytest.fail("eager Triton kernel entered during compilation"),
-    )
-
-    actual = norm(inputs)
-    expected = F.silu(F.group_norm(inputs, 2, norm.weight, norm.bias, norm.eps))
-    torch.testing.assert_close(actual, expected)
-
-
-def _small_image_vae() -> nn.Module:
-    from diffusers.models.autoencoders import AutoencoderKL
-
-    return AutoencoderKL(
-        in_channels=3,
-        out_channels=3,
-        down_block_types=("DownEncoderBlock2D", "DownEncoderBlock2D"),
-        up_block_types=("UpDecoderBlock2D", "UpDecoderBlock2D"),
-        block_out_channels=(32, 32),
-        layers_per_block=1,
-        latent_channels=4,
-        norm_num_groups=8,
-        sample_size=16,
-    )
-
-
-def _small_wan_vae(vae_cls: type[nn.Module] | None = None) -> nn.Module:
-    if vae_cls is None:
-        from diffusers.models.autoencoders import AutoencoderKLWan
-
-        vae_cls = AutoencoderKLWan
-
-    return vae_cls(
-        base_dim=8,
-        decoder_base_dim=8,
-        z_dim=4,
-        dim_mult=[1, 2],
-        num_res_blocks=1,
-        attn_scales=[],
-        temperal_downsample=[False],
-        latents_mean=[0.0] * 4,
-        latents_std=[1.0] * 4,
-        scale_factor_temporal=1,
-        scale_factor_spatial=2,
-    )
-
-
-@pytest.mark.core_model
-@hardware_test(res={"cuda": "L4"}, num_cards=1)
-@torch.no_grad()
-def test_image_vae_install_preserves_lossless_path_and_state_dict() -> None:
-    from vllm_omni.diffusion.vae_optimizations import clear_pipeline_vae_fast_path_caches
-    from vllm_omni.diffusion.vae_optimizations.image import (
-        FusedGroupNormSiLU,
-        FusedUpsample2xConv2d,
-        maybe_optimize_image_vae,
-    )
-
-    torch.manual_seed(0)
-    vae = _small_image_vae().to(device="cuda", dtype=torch.bfloat16).eval()
-    latents = torch.randn(1, 4, 8, 8, device="cuda", dtype=torch.bfloat16)
-    parameter_names = {name for name, _ in vae.named_parameters()}
-    state_dict = {name: value.clone() for name, value in vae.state_dict().items()}
-    reference = vae.decode(latents).sample
-
-    maybe_optimize_image_vae(vae)
-    maybe_optimize_image_vae(vae)
-
-    assert {name for name, _ in vae.named_parameters()} == parameter_names
-    vae.load_state_dict(state_dict, strict=True)
-    assert any(isinstance(module, FusedGroupNormSiLU) for module in vae.modules())
-    assert any(isinstance(module, FusedUpsample2xConv2d) for module in vae.modules())
-    assert torch.equal(vae.decode(latents).sample, reference)
-
-    with use_vae_fast_path(vae, True):
-        fast = vae.decode(latents).sample
-
-    torch.testing.assert_close(fast.float(), reference.float(), atol=0.1, rtol=0)
-    assert any(
-        buffer is not None
-        for module in vae.modules()
-        for name, buffer in module._buffers.items()
-        if name in {"_fused_weight", "_vllm_folded_value_weight", "_vllm_folded_value_bias"}
-    )
-    clear_pipeline_vae_fast_path_caches(_PipelineWithVae(vae))
-    assert all(
-        buffer is None
-        for module in vae.modules()
-        for name, buffer in module._buffers.items()
-        if name in {"_fused_weight", "_vllm_folded_value_weight", "_vllm_folded_value_bias"}
-    )
-    with use_vae_fast_path(vae, True):
-        torch.testing.assert_close(vae.decode(latents).sample.float(), reference.float(), atol=0.1, rtol=0)
-    assert torch.equal(vae.decode(latents).sample, reference)
-    assert vae.decoder.conv_in.weight.is_contiguous()
-
-
-@pytest.mark.core_model
-@hardware_test(res={"cuda": "L4"}, num_cards=1)
-@torch.no_grad()
-@pytest.mark.parametrize("dtype", [torch.float32, torch.bfloat16])
-def test_channels_last_group_norm_silu_matches_reference(dtype: torch.dtype) -> None:
-    from vllm_omni.diffusion.vae_optimizations.triton_group_norm_silu import group_norm_silu_4d
-
-    torch.manual_seed(0)
-    x = torch.randn(1, 64, 16, 16, device="cuda", dtype=dtype).contiguous(memory_format=torch.channels_last)
-    weight = torch.randn(64, device="cuda", dtype=dtype)
-    bias = torch.randn(64, device="cuda", dtype=dtype)
-
-    actual = group_norm_silu_4d(x, weight, bias, num_groups=8, eps=1e-5)
-    expected = F.silu(F.group_norm(x, 8, weight, bias))
-
-    assert actual is not None
-    atol, rtol = (1e-5, 1e-5) if dtype == torch.float32 else (7e-2, 2e-2)
-    torch.testing.assert_close(actual, expected, atol=atol, rtol=rtol)
-
-
-@pytest.mark.core_model
-@hardware_test(res={"cuda": "L4"}, num_cards=1)
-@torch.no_grad()
-@pytest.mark.parametrize(
-    "activation_dtype,affine_dtype,atol,rtol",
-    [
-        (torch.float32, torch.float32, 1e-5, 1e-5),
-        (torch.bfloat16, torch.float32, 1.5e-1, 3e-2),
-    ],
-)
-def test_wan_rmsnorm_silu_matches_reference(
-    activation_dtype: torch.dtype,
-    affine_dtype: torch.dtype,
-    atol: float,
-    rtol: float,
-) -> None:
-    from vllm_omni.diffusion.vae_optimizations.triton_wan_rmsnorm_silu import wan_rmsnorm_silu
-
-    torch.manual_seed(0)
-    x = torch.randn(1, 96, 3, 10, 14, device="cuda", dtype=activation_dtype).contiguous(
-        memory_format=torch.channels_last_3d
-    )
-    gamma = torch.randn(96, 1, 1, 1, device="cuda", dtype=affine_dtype)
-    bias = torch.randn_like(gamma)
-    normalized = F.normalize(x.float() if activation_dtype != torch.float32 else x, dim=1).to(activation_dtype)
-    expected = F.silu(normalized * 96**0.5 * gamma + bias)
-
-    actual = wan_rmsnorm_silu(x, gamma, bias)
-
-    assert actual is not None
-    assert actual.stride() == x.stride()
-    torch.testing.assert_close(actual, expected, atol=atol, rtol=rtol)
-
-
-@pytest.mark.core_model
-@hardware_test(res={"cuda": "L4"}, num_cards=1)
-@torch.no_grad()
-def test_wan_vae_install_preserves_lossless_path_and_state_dict() -> None:
-    from vllm_omni.diffusion.vae_optimizations.wan import (
-        FusedWanRMSNormSiLU,
-        maybe_optimize_wan_vae,
-    )
-
-    torch.manual_seed(0)
-    vae = _small_wan_vae().cuda().eval()
-    latents = torch.randn(1, 4, 1, 8, 8, device="cuda")
-    parameter_names = {name for name, _ in vae.named_parameters()}
-    state_dict = {name: value.clone() for name, value in vae.state_dict().items()}
-    reference = vae.decode(latents).sample
-
-    maybe_optimize_wan_vae(vae)
-    maybe_optimize_wan_vae(vae)
-
-    assert {name for name, _ in vae.named_parameters()} == parameter_names
-    vae.load_state_dict(state_dict, strict=True)
-    assert any(isinstance(module, FusedWanRMSNormSiLU) for module in vae.modules())
-    assert torch.equal(vae.decode(latents).sample, reference)
-
-    with use_vae_fast_path(vae, True):
-        fast = vae.decode(latents).sample
-
-    torch.testing.assert_close(fast, reference, atol=5e-4, rtol=1e-4)
-    assert torch.equal(vae.decode(latents).sample, reference)
-
-
-@pytest.mark.core_model
-@pytest.mark.cpu
-@pytest.mark.parametrize("parallel_mode", ["auto", "spatial_shard_height", "spatial_shard_width"])
-def test_wan_fast_path_skips_spatial_shard_mode(monkeypatch: pytest.MonkeyPatch, parallel_mode: str) -> None:
-    import vllm_omni.diffusion.vae_optimizations.wan as wan_optimizations
-
-    monkeypatch.setattr(wan_optimizations, "_HAS_TRITON", True)
-    vae = _small_wan_vae()
-    vae.distributed_executor = _DistributedExecutorConfig(parallel_mode)
-
-    wan_optimizations.maybe_optimize_wan_vae(vae)
-
-    assert not any(isinstance(module, wan_optimizations.FusedWanRMSNormSiLU) for module in vae.modules())
-
-
-@pytest.mark.core_model
-@hardware_test(res={"cuda": "L4"}, num_cards=1)
-@torch.no_grad()
-@pytest.mark.parametrize(
-    ("precision", "decode_dtype"),
-    [("fp32", torch.float32), ("fp16", torch.float16), ("bf16", torch.bfloat16)],
-)
-def test_wan_decode_precision_keeps_encode_fp32(precision: str, decode_dtype: torch.dtype) -> None:
-    from vllm_omni.diffusion.distributed.autoencoders.autoencoder_kl_wan import DistributedAutoencoderKLWan
-    from vllm_omni.diffusion.vae_optimizations.precision import configure_wan_decode_precision
-
-    torch.manual_seed(0)
-    vae = _small_wan_vae(DistributedAutoencoderKLWan).to(device="cuda", dtype=torch.bfloat16).eval()
-    configure_wan_decode_precision(vae, precision)
-
-    assert next(vae.encoder.parameters()).dtype == torch.float32
-    assert next(vae.quant_conv.parameters()).dtype == torch.float32
-    assert next(vae.decoder.parameters()).dtype == decode_dtype
-    assert next(vae.post_quant_conv.parameters()).dtype == decode_dtype
-
-    _, grid_spec = vae.tile_split(torch.randn(1, 4, 1, 8, 8, device="cuda", dtype=decode_dtype))
-    assert grid_spec.output_dtype == decode_dtype
-
-    encoded = vae.encode(torch.randn(1, 3, 1, 16, 16, device="cuda", dtype=torch.bfloat16))
-    decoded = vae.decode(torch.randn(1, 4, 1, 8, 8, device="cuda", dtype=torch.float32))
-
-    assert encoded.latent_dist.parameters.dtype == torch.float32
-    assert decoded.sample.dtype == decode_dtype

tests/entrypoints/test_serve.py

@@ -76,17 +76,6 @@ def test_serve_parser_accepts_four_way_cfg_parallelism() -> None:
     assert args.cfg_parallel_size == 4
 
 
-def test_serve_parser_accepts_wan_decode_precision() -> None:
-    parser = TrackingArgumentParser()
-    subparsers = parser.add_subparsers(dest="subcommand")
-    OmniServeCommand().subparser_init(subparsers)
-
-    args = parser.parse_args(["serve", "fake-model", "--omni", "--vae-decode-precision", "bf16"])
-
-    assert args.vae_decode_precision == "bf16"
-    assert args.get_explicit_kwargs_dict()["vae_decode_precision"] == "bf16"
-
-
 def _make_headless_args(**kwargs) -> TrackingNamespace:
     defaults = {
         "model": "fake-model",

vllm_omni/config/omni_config.py

@@ -573,7 +573,6 @@ class _DiffusionConfigProjection:
     fa_deterministic: bool = False
     vae_use_slicing: bool = False
     vae_use_tiling: bool = False
-    vae_decode_precision: Literal["fp32", "fp16", "bf16"] | None = None
     mask_strategy_file_path: str | None = None
     skip_time_steps: int = 15
     VSA_sparsity: float = 0.0

vllm_omni/config/stage_config.py

@@ -401,7 +401,6 @@ class StageDeployConfig:
     step_execution: bool | None = None
     vae_use_slicing: bool | None = None
     vae_use_tiling: bool | None = None
-    vae_decode_precision: str | None = None
     boundary_ratio: float | None = None
     flow_shift: float | None = None
     diffusion_kv_cache_dtype: str | None = None

vllm_omni/diffusion/data.py

@@ -711,9 +711,6 @@ class OmniDiffusionConfig:
     # VAE memory optimization parameters
     vae_use_slicing: bool = False
     vae_use_tiling: bool = False
-    # Optional Wan decode-only precision. When set, the VAE encoder remains
-    # FP32 while decoder-side modules use the selected precision.
-    vae_decode_precision: str | None = None
 
     # STA (Sliding Tile Attention) parameters
     mask_strategy_file_path: str | None = None
@@ -926,11 +923,6 @@ class OmniDiffusionConfig:
         )
 
     def __post_init__(self):
-        if self.vae_decode_precision not in (None, "fp32", "fp16", "bf16"):
-            raise ValueError(
-                "vae_decode_precision must be one of {'fp32', 'fp16', 'bf16'} or None, "
-                f"got {self.vae_decode_precision!r}"
-            )
         if self.diffusion_compile_granularity not in {"regional", "full"}:
             raise ValueError(
                 "diffusion_compile_granularity must be 'regional' or 'full', "

vllm_omni/diffusion/distributed/autoencoders/autoencoder_kl_wan.py

@@ -18,13 +18,26 @@ from vllm_omni.diffusion.distributed.autoencoders.distributed_vae_executor impor
     GridSpec,
     TileTask,
 )
+from vllm_omni.diffusion.distributed.autoencoders.wan_vae_data_movement import (
+    install_wan_vae_data_movement,
+)
 from vllm_omni.platforms import current_omni_platform
 
 logger = init_logger(__name__)
 
 
+def prepare_pipeline_wan_vae_data_movement(pipeline: torch.nn.Module) -> None:
+    """Install Wan data-movement methods before offload or compilation setup."""
+    from vllm_omni.diffusion.offloader.module_collector import ModuleDiscovery
+
+    for vae in ModuleDiscovery.discover(pipeline).vaes:
+        if isinstance(vae, OmniAutoencoderKLWan):
+            install_wan_vae_data_movement(vae)
+
+
 class OmniAutoencoderKLWan(AutoencoderKLWan):
     def _execution_context(self):
+        install_wan_vae_data_movement(self)
         try:
             first_param = next(self.parameters())
         except StopIteration:
@@ -106,7 +119,7 @@ class DistributedAutoencoderKLWan(OmniAutoencoderKLWan, DistributedVaeMixin):
             split_dims=(3, 4),
             grid_shape=(tiletask_list[-1].grid_coord[0] + 1, tiletask_list[-1].grid_coord[1] + 1),
             tile_spec=tile_spec,
-            output_dtype=getattr(self, "_vllm_decode_dtype", self.dtype),
+            output_dtype=self.dtype,
         )
         return tiletask_list, grid_spec
 

vllm_omni/diffusion/distributed/autoencoders/wan_vae_data_movement.py

@@ -0,0 +1,302 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Guarded Wan VAE decoder data-movement fast paths.
+
+This module adapts the lossless Wan decoder optimizations from SGLang while
+keeping Diffusers' parameter names and module topology unchanged. Every
+optimized method falls back to the Diffusers 0.38.0 operation order when the
+kernel, device, layout, dtype, autograd state, or module type is unsupported.
+"""
+
+from __future__ import annotations
+
+from types import MethodType
+from typing import Any
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from diffusers.models.autoencoders.autoencoder_kl_wan import (
+    CACHE_T,
+    DupUp3D,
+    WanCausalConv3d,
+    WanDecoder3d,
+    WanResample,
+    WanResidualBlock,
+    WanResidualUpBlock,
+)
+
+from vllm_omni.diffusion.distributed.autoencoders import wan_spatial_shard
+
+try:
+    from vllm_omni.diffusion.kernels.wan_vae_data_movement import (
+        cat_pad_5d,
+        dup_up3d_add,
+    )
+except ImportError:  # pragma: no cover - Triton is optional on non-GPU installs.
+    cat_pad_5d = None
+    dup_up3d_add = None
+
+_INSTALL_ATTR = "_vllm_omni_wan_data_movement_installed"
+_DYNAMIC_SPATIAL_CONV_ATTR = "_vllm_omni_dynamic_spatial_shard_conv"
+
+
+def _cache_payload(cache: Any) -> torch.Tensor | None:
+    return cache if isinstance(cache, torch.Tensor) else None
+
+
+def _is_dynamic_spatial_conv(conv: nn.Module) -> bool:
+    return bool(getattr(conv, _DYNAMIC_SPATIAL_CONV_ATTR, False))
+
+
+def _spatial_shard_is_active() -> bool:
+    return wan_spatial_shard._SPATIAL_SHARD_CONTEXT.get() is not None
+
+
+def _causal_padding(conv: nn.Module) -> tuple[int, ...]:
+    if _is_dynamic_spatial_conv(conv):
+        return tuple(conv._source_padding)
+    return tuple(conv._padding)
+
+
+def _can_use_fused_cache_path(conv: nn.Module, x: torch.Tensor) -> bool:
+    return bool(
+        cat_pad_5d is not None
+        and (type(conv) is WanCausalConv3d or _is_dynamic_spatial_conv(conv))
+        and x.dim() == 5
+        and x.is_cuda
+        and not torch.is_grad_enabled()
+        and not torch.compiler.is_compiling()
+        and not _spatial_shard_is_active()
+    )
+
+
+def _run_cached_causal_conv(
+    conv: nn.Module,
+    x: torch.Tensor,
+    cache_list: list[Any],
+    index: int,
+) -> torch.Tensor:
+    """Run a causal conv and refresh its feature-cache slot."""
+    cache = cache_list[index]
+    is_repeat_marker = isinstance(cache, str)
+    payload = None if is_repeat_marker else _cache_payload(cache)
+    if _can_use_fused_cache_path(conv, x) and (
+        payload is None or (payload.device == x.device and payload.dtype == x.dtype)
+    ):
+        pair = cat_pad_5d(x, payload, _causal_padding(conv), keep_cache_frames=CACHE_T)
+        if pair is not None:
+            conv_input, cache_list[index] = pair
+            return nn.Conv3d.forward(conv, conv_input)
+
+    cache_x = x[:, :, -CACHE_T:, :, :].clone()
+    if cache_x.shape[2] < CACHE_T and payload is not None:
+        cache_x = torch.cat(
+            [payload[:, :, -1, :, :].unsqueeze(2).to(cache_x.device), cache_x],
+            dim=2,
+        )
+    elif cache_x.shape[2] < CACHE_T and is_repeat_marker:
+        cache_x = torch.cat([torch.zeros_like(cache_x), cache_x], dim=2)
+
+    out = conv(x) if payload is None else conv(x, payload)
+    cache_list[index] = cache_x
+    return out
+
+
+def _causal_conv_forward(self: WanCausalConv3d, x: torch.Tensor, cache_x: torch.Tensor | None = None):
+    padding = list(self._padding)
+    if (
+        any(padding)
+        and _can_use_fused_cache_path(self, x)
+        and (cache_x is None or (cache_x.device == x.device and cache_x.dtype == x.dtype))
+    ):
+        conv_input = cat_pad_5d(x, cache_x, padding)
+        if conv_input is not None:
+            return nn.Conv3d.forward(self, conv_input)
+
+    if cache_x is not None and padding[4] > 0:
+        cache_x = cache_x.to(x.device)
+        x = torch.cat([cache_x, x], dim=2)
+        padding[4] -= cache_x.shape[2]
+    x = F.pad(x, padding)
+    return nn.Conv3d.forward(self, x)
+
+
+def _resample_forward(
+    self: WanResample,
+    x: torch.Tensor,
+    feat_cache: list[Any] | None = None,
+    feat_idx: list[int] | None = None,
+):
+    if feat_idx is None:
+        feat_idx = [0]
+    batch, channels, frames, height, width = x.size()
+    if self.mode == "upsample3d" and feat_cache is not None:
+        index = feat_idx[0]
+        if feat_cache[index] is None:
+            feat_cache[index] = "Rep"
+            feat_idx[0] += 1
+        else:
+            x = _run_cached_causal_conv(self.time_conv, x, feat_cache, index)
+            feat_idx[0] += 1
+            x = x.reshape(batch, 2, channels, frames, height, width)
+            x = torch.stack((x[:, 0], x[:, 1]), 3)
+            x = x.reshape(batch, channels, frames * 2, height, width)
+
+    frames = x.shape[2]
+    x = x.permute(0, 2, 1, 3, 4).reshape(batch * frames, channels, height, width)
+    x = self.resample(x)
+    x = x.view(batch, frames, x.size(1), x.size(2), x.size(3)).permute(0, 2, 1, 3, 4)
+
+    if self.mode == "downsample3d" and feat_cache is not None:
+        index = feat_idx[0]
+        if feat_cache[index] is None:
+            feat_cache[index] = x.clone()
+            feat_idx[0] += 1
+        else:
+            cache_x = x[:, :, -1:, :, :].clone()
+            x = self.time_conv(torch.cat([feat_cache[index][:, :, -1:, :, :], x], dim=2))
+            feat_cache[index] = cache_x
+            feat_idx[0] += 1
+    return x
+
+
+def _residual_block_forward(
+    self: WanResidualBlock,
+    x: torch.Tensor,
+    feat_cache: list[Any] | None = None,
+    feat_idx: list[int] | None = None,
+):
+    if feat_idx is None:
+        feat_idx = [0]
+    residual = self.conv_shortcut(x)
+    x = self.nonlinearity(self.norm1(x))
+    if feat_cache is not None:
+        index = feat_idx[0]
+        x = _run_cached_causal_conv(self.conv1, x, feat_cache, index)
+        feat_idx[0] += 1
+    else:
+        x = self.conv1(x)
+
+    x = self.dropout(self.nonlinearity(self.norm2(x)))
+    if feat_cache is not None:
+        index = feat_idx[0]
+        x = _run_cached_causal_conv(self.conv2, x, feat_cache, index)
+        feat_idx[0] += 1
+    else:
+        x = self.conv2(x)
+    return x + residual
+
+
+def _residual_up_block_forward(
+    self: WanResidualUpBlock,
+    x: torch.Tensor,
+    feat_cache: list[Any] | None = None,
+    feat_idx: list[int] | None = None,
+    first_chunk: bool = False,
+):
+    if feat_idx is None:
+        feat_idx = [0]
+    shortcut_source = x.clone()
+    for resnet in self.resnets:
+        if feat_cache is None:
+            x = resnet(x)
+        else:
+            x = resnet(x, feat_cache=feat_cache, feat_idx=feat_idx)
+
+    if self.upsampler is not None:
+        if feat_cache is None:
+            x = self.upsampler(x)
+        else:
+            x = self.upsampler(x, feat_cache=feat_cache, feat_idx=feat_idx)
+
+    shortcut = self.avg_shortcut
+    if shortcut is None:
+        return x
+    if (
+        dup_up3d_add is not None
+        and type(shortcut) is DupUp3D
+        and x.is_cuda
+        and shortcut_source.is_cuda
+        and x.dtype == shortcut_source.dtype
+        and not torch.is_grad_enabled()
+        and not torch.compiler.is_compiling()
+        and not _spatial_shard_is_active()
+    ):
+        fused = dup_up3d_add(
+            x,
+            shortcut_source,
+            shortcut.factor_t,
+            shortcut.factor_s,
+            shortcut.repeats,
+            first_chunk,
+        )
+        if fused is not None:
+            return fused
+    return x + shortcut(shortcut_source, first_chunk=first_chunk)
+
+
+def _decoder_forward(
+    self: WanDecoder3d,
+    x: torch.Tensor,
+    feat_cache: list[Any] | None = None,
+    feat_idx: list[int] | None = None,
+    first_chunk: bool = False,
+):
+    if feat_idx is None:
+        feat_idx = [0]
+    if feat_cache is not None:
+        index = feat_idx[0]
+        x = _run_cached_causal_conv(self.conv_in, x, feat_cache, index)
+        feat_idx[0] += 1
+    else:
+        x = self.conv_in(x)
+
+    x = self.mid_block(x, feat_cache=feat_cache, feat_idx=feat_idx)
+    for up_block in self.up_blocks:
+        x = up_block(x, feat_cache=feat_cache, feat_idx=feat_idx, first_chunk=first_chunk)
+
+    x = self.nonlinearity(self.norm_out(x))
+    if feat_cache is not None:
+        index = feat_idx[0]
+        x = _run_cached_causal_conv(self.conv_out, x, feat_cache, index)
+        feat_idx[0] += 1
+    else:
+        x = self.conv_out(x)
+    return x
+
+
+def install_wan_vae_data_movement(vae: nn.Module) -> bool:
+    """Install guarded fast paths on a Diffusers Wan decoder once."""
+    if getattr(vae, _INSTALL_ATTR, False):
+        return True
+    if cat_pad_5d is None or dup_up3d_add is None:
+        return False
+
+    decoder = getattr(vae, "decoder", None)
+    if type(decoder) is not WanDecoder3d:
+        return False
+
+    decoder.forward = MethodType(_decoder_forward, decoder)
+    patched = 0
+    for module in decoder.modules():
+        module_type = type(module)
+        if module_type is WanCausalConv3d:
+            module.forward = MethodType(_causal_conv_forward, module)
+            patched += 1
+        elif _is_dynamic_spatial_conv(module):
+            # The dynamic wrapper owns halo exchange while a spatial request is
+            # active. Its direct path can still use the lossless fused cache
+            # movement through ``_run_cached_causal_conv``.
+            patched += 1
+        elif module_type is WanResidualBlock:
+            module.forward = MethodType(_residual_block_forward, module)
+        elif module_type is WanResidualUpBlock:
+            module.forward = MethodType(_residual_up_block_forward, module)
+        elif module_type is WanResample:
+            module.forward = MethodType(_resample_forward, module)
+
+    if patched == 0:
+        return False
+    setattr(vae, _INSTALL_ATTR, True)
+    return True

vllm_omni/diffusion/kernels/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

vllm_omni/diffusion/kernels/wan_vae_data_movement.py

@@ -0,0 +1,405 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Bit-exact data-movement kernels for the Wan causal VAE decoder.
+
+Adapted from SGLang PR #34125:
+https://github.com/sgl-project/sglang/pull/34125
+
+The kernels in this module only move values, fill zeros, or perform the same
+two-operand addition as the PyTorch reference path. They preserve the input
+layout instead of selecting a new Conv3d layout, so enabling them does not
+change the decoder's cuDNN accumulation order.
+"""
+
+from __future__ import annotations
+
+import torch
+import triton
+import triton.language as tl
+
+_MAX_INT32 = 2**31 - 1
+_SUPPORTED_DTYPES = (torch.float16, torch.bfloat16, torch.float32)
+
+
+@triton.jit
+def _cat_pad_5d_kernel(
+    x_ptr,
+    cache_ptr,
+    out_ptr,
+    keep_ptr,
+    total,
+    channels,
+    frames,
+    height,
+    width,
+    cache_frames,
+    out_frames,
+    out_height,
+    out_width,
+    zero_front_frames,
+    pad_height,
+    pad_width,
+    stride_x_batch,
+    stride_x_channel,
+    stride_x_frame,
+    stride_x_height,
+    stride_x_width,
+    stride_cache_batch,
+    stride_cache_channel,
+    stride_cache_frame,
+    stride_cache_height,
+    stride_cache_width,
+    has_cache: tl.constexpr,
+    keep_frames_const: tl.constexpr,
+    channels_inner: tl.constexpr,
+    index_64bit: tl.constexpr,
+    block_size_const: tl.constexpr,
+):
+    if index_64bit:
+        offsets = tl.program_id(0).to(tl.int64) * block_size_const + tl.arange(0, block_size_const).to(tl.int64)
+    else:
+        offsets = tl.program_id(0) * block_size_const + tl.arange(0, block_size_const)
+    mask = offsets < total
+
+    if channels_inner:
+        out_channel = offsets % channels
+        remaining = offsets // channels
+        out_width_idx = remaining % out_width
+        remaining = remaining // out_width
+        out_height_idx = remaining % out_height
+        remaining = remaining // out_height
+        out_frame = remaining % out_frames
+        out_batch = remaining // out_frames
+    else:
+        out_width_idx = offsets % out_width
+        remaining = offsets // out_width
+        out_height_idx = remaining % out_height
+        remaining = remaining // out_height
+        out_frame = remaining % out_frames
+        remaining = remaining // out_frames
+        out_channel = remaining % channels
+        out_batch = remaining // channels
+
+    input_width = out_width_idx - pad_width
+    input_height = out_height_idx - pad_height
+    input_frame = out_frame - zero_front_frames
+
+    spatial_valid = (input_width >= 0) & (input_width < width) & (input_height >= 0) & (input_height < height)
+    from_cache = spatial_valid & (input_frame >= 0) & (input_frame < cache_frames)
+    from_x = spatial_valid & (input_frame >= cache_frames) & (input_frame < cache_frames + frames)
+
+    x_frame = input_frame - cache_frames
+    x_offset = (
+        out_batch * stride_x_batch
+        + out_channel * stride_x_channel
+        + x_frame * stride_x_frame
+        + input_height * stride_x_height
+        + input_width * stride_x_width
+    )
+    values = tl.load(x_ptr + x_offset, mask=mask & from_x, other=0.0)
+    if has_cache:
+        cache_offset = (
+            out_batch * stride_cache_batch
+            + out_channel * stride_cache_channel
+            + input_frame * stride_cache_frame
+            + input_height * stride_cache_height
+            + input_width * stride_cache_width
+        )
+        cache_values = tl.load(cache_ptr + cache_offset, mask=mask & from_cache, other=0.0)
+        values = tl.where(from_cache, cache_values, values)
+    tl.store(out_ptr + offsets, values, mask=mask)
+
+    if keep_frames_const > 0:
+        keep_frame = out_frame - (out_frames - keep_frames_const)
+        keep_mask = mask & spatial_valid & (keep_frame >= 0)
+        if channels_inner:
+            keep_offset = (
+                ((out_batch * keep_frames_const + keep_frame) * height + input_height) * width + input_width
+            ) * channels + out_channel
+        else:
+            keep_offset = (
+                ((out_batch * channels + out_channel) * keep_frames_const + keep_frame) * height + input_height
+            ) * width + input_width
+        tl.store(keep_ptr + keep_offset, values, mask=keep_mask)
+
+
+def _uses_channels_last_3d(tensor: torch.Tensor) -> bool:
+    return tensor.shape[1] > 1 and tensor.stride(1) == 1
+
+
+def cat_pad_5d(
+    x: torch.Tensor,
+    cache_x: torch.Tensor | None,
+    padding: list[int] | tuple[int, ...],
+    keep_cache_frames: int = 0,
+) -> torch.Tensor | tuple[torch.Tensor, torch.Tensor] | None:
+    """Fuse temporal concat, constant padding, and optional cache refresh.
+
+    ``padding`` follows the Wan causal Conv3d convention
+    ``(w_left, w_right, h_top, h_bottom, t_front, t_back)``. The output uses
+    the same dense layout that the reference ``cat``/``pad`` path selects.
+    Unsupported inputs return ``None`` so callers can execute the reference
+    implementation.
+    """
+    if len(padding) != 6:
+        return None
+    pad_width_left, pad_width_right, pad_height_top, pad_height_bottom, pad_front, pad_back = padding
+    if any(value < 0 for value in padding):
+        return None
+    if pad_width_left != pad_width_right or pad_height_top != pad_height_bottom or pad_back != 0:
+        return None
+    if x.dim() != 5 or not x.is_cuda or x.dtype not in _SUPPORTED_DTYPES:
+        return None
+    if keep_cache_frames < 0:
+        return None
+
+    channels_inner = _uses_channels_last_3d(x)
+    cache_frames = 0
+    if cache_x is not None:
+        if (
+            cache_x.dim() != 5
+            or cache_x.dtype != x.dtype
+            or cache_x.device != x.device
+            or cache_x.shape[0] != x.shape[0]
+            or cache_x.shape[1] != x.shape[1]
+            or cache_x.shape[3:] != x.shape[3:]
+            or _uses_channels_last_3d(cache_x) != channels_inner
+        ):
+            return None
+        cache_frames = cache_x.shape[2]
+
+    zero_front_frames = pad_front - cache_frames
+    if zero_front_frames < 0:
+        return None
+
+    batch, channels, frames, height, width = x.shape
+    out_frames = pad_front + frames
+    out_height = height + 2 * pad_height_top
+    out_width = width + 2 * pad_width_left
+    keep_frames = min(keep_cache_frames, out_frames)
+    memory_format = torch.channels_last_3d if channels_inner else torch.contiguous_format
+    out = torch.empty(
+        (batch, channels, out_frames, out_height, out_width),
+        device=x.device,
+        dtype=x.dtype,
+        memory_format=memory_format,
+    )
+    total = out.numel()
+    if total == 0 or total > _MAX_INT32 * 4:
+        return None
+
+    if keep_frames:
+        keep_arg = torch.empty(
+            (batch, channels, keep_frames, height, width),
+            device=x.device,
+            dtype=x.dtype,
+            memory_format=memory_format,
+        )
+    else:
+        keep_arg = out
+
+    if cache_x is None:
+        cache_arg = x
+        cache_strides = (0, 0, 0, 0, 0)
+    else:
+        cache_arg = cache_x
+        cache_strides = cache_x.stride()
+
+    block_size = 512
+    grid = (triton.cdiv(total, block_size),)
+    with torch.get_device_module().device(x.device):
+        _cat_pad_5d_kernel[grid](
+            x,
+            cache_arg,
+            out,
+            keep_arg,
+            total,
+            channels,
+            frames,
+            height,
+            width,
+            cache_frames,
+            out_frames,
+            out_height,
+            out_width,
+            zero_front_frames,
+            pad_height_top,
+            pad_width_left,
+            *x.stride(),
+            *cache_strides,
+            has_cache=cache_x is not None,
+            keep_frames_const=keep_frames,
+            channels_inner=channels_inner,
+            index_64bit=total >= _MAX_INT32,
+            block_size_const=block_size,
+        )
+
+    if keep_cache_frames:
+        return out, keep_arg
+    return out
+
+
+@triton.jit
+def _dup_up3d_add_kernel(
+    main_ptr,
+    source_ptr,
+    out_ptr,
+    total,
+    out_channels,
+    out_frames,
+    out_height,
+    out_width,
+    frame_offset,
+    stride_main_batch,
+    stride_main_channel,
+    stride_main_frame,
+    stride_main_height,
+    stride_main_width,
+    stride_source_batch,
+    stride_source_channel,
+    stride_source_frame,
+    stride_source_height,
+    stride_source_width,
+    stride_out_batch,
+    stride_out_channel,
+    stride_out_frame,
+    stride_out_height,
+    stride_out_width,
+    factor_temporal_const: tl.constexpr,
+    factor_spatial_const: tl.constexpr,
+    repeats_const: tl.constexpr,
+    channels_inner: tl.constexpr,
+    index_64bit: tl.constexpr,
+    block_size_const: tl.constexpr,
+):
+    if index_64bit:
+        offsets = tl.program_id(0).to(tl.int64) * block_size_const + tl.arange(0, block_size_const).to(tl.int64)
+    else:
+        offsets = tl.program_id(0) * block_size_const + tl.arange(0, block_size_const)
+    mask = offsets < total
+
+    if channels_inner:
+        out_channel = offsets % out_channels
+        remaining = offsets // out_channels
+        out_width_idx = remaining % out_width
+        remaining = remaining // out_width
+        out_height_idx = remaining % out_height
+        remaining = remaining // out_height
+        out_frame = remaining % out_frames
+        out_batch = remaining // out_frames
+    else:
+        out_width_idx = offsets % out_width
+        remaining = offsets // out_width
+        out_height_idx = remaining % out_height
+        remaining = remaining // out_height
+        out_frame = remaining % out_frames
+        remaining = remaining // out_frames
+        out_channel = remaining % out_channels
+        out_batch = remaining // out_channels
+
+    unsliced_frame = out_frame + frame_offset
+    source_frame = unsliced_frame // factor_temporal_const
+    temporal_remainder = unsliced_frame % factor_temporal_const
+    source_height = out_height_idx // factor_spatial_const
+    height_remainder = out_height_idx % factor_spatial_const
+    source_width = out_width_idx // factor_spatial_const
+    width_remainder = out_width_idx % factor_spatial_const
+    repeated_channel = (
+        (out_channel * factor_temporal_const + temporal_remainder) * factor_spatial_const + height_remainder
+    ) * factor_spatial_const + width_remainder
+    source_channel = repeated_channel // repeats_const
+
+    main_offset = (
+        out_batch * stride_main_batch
+        + out_channel * stride_main_channel
+        + out_frame * stride_main_frame
+        + out_height_idx * stride_main_height
+        + out_width_idx * stride_main_width
+    )
+    source_offset = (
+        out_batch * stride_source_batch
+        + source_channel * stride_source_channel
+        + source_frame * stride_source_frame
+        + source_height * stride_source_height
+        + source_width * stride_source_width
+    )
+    out_offset = (
+        out_batch * stride_out_batch
+        + out_channel * stride_out_channel
+        + out_frame * stride_out_frame
+        + out_height_idx * stride_out_height
+        + out_width_idx * stride_out_width
+    )
+    main_value = tl.load(main_ptr + main_offset, mask=mask, other=0.0)
+    source_value = tl.load(source_ptr + source_offset, mask=mask, other=0.0)
+    value = main_value.to(tl.float32) + source_value.to(tl.float32)
+    tl.store(out_ptr + out_offset, value, mask=mask)
+
+
+def dup_up3d_add(
+    main: torch.Tensor,
+    source: torch.Tensor,
+    factor_temporal: int,
+    factor_spatial: int,
+    repeats: int,
+    drop_first_frames: bool,
+) -> torch.Tensor | None:
+    """Evaluate ``main + DupUp3D(source)`` without materializing the shortcut."""
+    if main.dim() != 5 or source.dim() != 5:
+        return None
+    if factor_temporal <= 0 or factor_spatial <= 0 or repeats <= 0:
+        return None
+    if factor_temporal & (factor_temporal - 1) or factor_spatial & (factor_spatial - 1):
+        return None
+    if repeats & (repeats - 1):
+        return None
+    if not main.is_cuda or not source.is_cuda:
+        return None
+    if main.dtype != source.dtype or main.dtype not in _SUPPORTED_DTYPES or main.device != source.device:
+        return None
+
+    batch, source_channels, frames, height, width = source.shape
+    frame_offset = factor_temporal - 1 if drop_first_frames else 0
+    numerator = source_channels * repeats
+    denominator = factor_temporal * factor_spatial * factor_spatial
+    if numerator % denominator:
+        return None
+    expected_shape = (
+        batch,
+        numerator // denominator,
+        frames * factor_temporal - frame_offset,
+        height * factor_spatial,
+        width * factor_spatial,
+    )
+    if tuple(main.shape) != expected_shape:
+        return None
+
+    out = torch.empty_like(main)
+    total = out.numel()
+    if total == 0 or total > _MAX_INT32 * 4:
+        return None
+
+    block_size = 512
+    grid = (triton.cdiv(total, block_size),)
+    with torch.get_device_module().device(main.device):
+        _dup_up3d_add_kernel[grid](
+            main,
+            source,
+            out,
+            total,
+            expected_shape[1],
+            expected_shape[2],
+            expected_shape[3],
+            expected_shape[4],
+            frame_offset,
+            *main.stride(),
+            *source.stride(),
+            *out.stride(),
+            factor_temporal_const=factor_temporal,
+            factor_spatial_const=factor_spatial,
+            repeats_const=repeats,
+            channels_inner=out.stride(1) == 1 and expected_shape[1] > 1,
+            index_64bit=total >= _MAX_INT32,
+            block_size_const=block_size,
+        )
+    return out

vllm_omni/diffusion/models/helios/pipeline_helios.py

@@ -21,6 +21,7 @@ from transformers import AutoConfig, AutoTokenizer, UMT5EncoderModel
 from vllm.model_executor.models.utils import AutoWeightsLoader
 
 from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
+from vllm_omni.diffusion.distributed.autoencoders.wan_vae_data_movement import install_wan_vae_data_movement
 from vllm_omni.diffusion.distributed.cfg_parallel import CFGParallelMixin
 from vllm_omni.diffusion.distributed.utils import get_local_device
 from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
@@ -215,6 +216,7 @@ class HeliosPipeline(
         self.vae = AutoencoderKLWan.from_pretrained(
             model, subfolder="vae", torch_dtype=torch.float32, local_files_only=local_files_only
         ).to(self.device)
+        install_wan_vae_data_movement(self.vae)
 
         transformer_config = load_transformer_config(model, "transformer", local_files_only)
         self.transformer = create_transformer_from_config(

vllm_omni/diffusion/models/lingbot_video/pipeline_lingbot_video.py

@@ -19,6 +19,7 @@ from torch import nn
 from transformers import Qwen3VLForConditionalGeneration, Qwen3VLProcessor
 
 from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
+from vllm_omni.diffusion.distributed.autoencoders.wan_vae_data_movement import install_wan_vae_data_movement
 from vllm_omni.diffusion.distributed.utils import get_local_device
 from vllm_omni.diffusion.models.interface import (
     SupportImageInput,
@@ -374,6 +375,7 @@ class LingBotVideoPipeline(
             torch_dtype=vae_dtype,
             local_files_only=local_files_only,
         ).to(self.device)
+        install_wan_vae_data_movement(self.vae)
         self.scheduler = FlowUniPCMultistepScheduler.from_pretrained(
             model,
             subfolder=scheduler_subfolder,

vllm_omni/diffusion/vae_optimizations/__init__.py

@@ -1,19 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Optional VAE decoder optimizations."""
-
-from vllm_omni.diffusion.vae_optimizations.gate import (
-    use_pipeline_vae_fast_path,
-    use_vae_fast_path,
-)
-from vllm_omni.diffusion.vae_optimizations.optimize import (
-    clear_pipeline_vae_fast_path_caches,
-    optimize_pipeline_vaes,
-)
-
-__all__ = [
-    "clear_pipeline_vae_fast_path_caches",
-    "optimize_pipeline_vaes",
-    "use_pipeline_vae_fast_path",
-    "use_vae_fast_path",
-]

vllm_omni/diffusion/vae_optimizations/gate.py

@@ -1,67 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Decode-scoped gates for optional, non-bit-exact VAE fast paths."""
-
-from __future__ import annotations
-
-from collections.abc import Iterator
-from contextlib import ExitStack, contextmanager
-from weakref import WeakKeyDictionary
-
-from torch import nn
-
-from vllm_omni.diffusion.offloader.module_collector import ModuleDiscovery
-
-
-class VaeFastPathGate:
-    """Mutable flag shared by wrappers installed on one VAE instance."""
-
-    __slots__ = ("enabled",)
-
-    def __init__(self) -> None:
-        self.enabled = False
-
-
-_VAE_FAST_PATH_GATES: WeakKeyDictionary[nn.Module, VaeFastPathGate] = WeakKeyDictionary()
-
-
-def register_vae_fast_path_gate(vae: nn.Module, gate: VaeFastPathGate) -> None:
-    """Associate an installed fast path with its owning VAE."""
-
-    _VAE_FAST_PATH_GATES[vae] = gate
-
-
-@contextmanager
-def use_vae_fast_path(vae: nn.Module, enabled: bool) -> Iterator[None]:
-    """Enable an installed fast path for one decode and always restore it."""
-
-    gate = _VAE_FAST_PATH_GATES.get(vae)
-    if gate is None:
-        yield
-        return
-
-    previous_enabled = gate.enabled
-    gate.enabled = enabled
-    try:
-        yield
-    finally:
-        gate.enabled = previous_enabled
-
-
-@contextmanager
-def use_pipeline_vae_fast_path(pipeline: nn.Module, enabled: bool) -> Iterator[None]:
-    """Apply one request's quality choice to every declared VAE component."""
-
-    vaes = ModuleDiscovery.discover(pipeline).vaes
-    with ExitStack() as stack:
-        for vae in vaes:
-            stack.enter_context(use_vae_fast_path(vae, enabled))
-        yield
-
-
-__all__ = [
-    "VaeFastPathGate",
-    "register_vae_fast_path_gate",
-    "use_pipeline_vae_fast_path",
-    "use_vae_fast_path",
-]

vllm_omni/diffusion/vae_optimizations/image.py

@@ -1,354 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Quality-gated CUDA fast paths for Diffusers-style 2D KL VAE decoders."""
-
-from __future__ import annotations
-
-from types import MethodType
-from typing import Any
-
-import torch
-import torch.nn.functional as F
-from torch import nn
-from vllm.logger import init_logger
-
-from vllm_omni.diffusion.vae_optimizations.gate import (
-    VaeFastPathGate,
-    register_vae_fast_path_gate,
-)
-
-logger = init_logger(__name__)
-
-try:
-    from vllm_omni.diffusion.vae_optimizations.triton_group_norm_silu import (
-        group_norm_silu_4d,
-        group_norm_silu_rows,
-    )
-
-    _HAS_TRITON = True
-except ImportError:  # pragma: no cover - exercised on builds without Triton
-    _HAS_TRITON = False
-
-
-class FusedGroupNormSiLU(nn.Module):
-    """Quality-gated GroupNorm + SiLU with an exact eager fallback."""
-
-    def __init__(self, norm: nn.GroupNorm, gate: VaeFastPathGate) -> None:
-        super().__init__()
-        self.num_groups = norm.num_groups
-        self.num_channels = norm.num_channels
-        self.eps = norm.eps
-        self.affine = norm.affine
-        # Register parameters directly so state-dict FQNs do not change.
-        self.weight = norm.weight
-        self.bias = norm.bias
-        self._vllm_gate = gate
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        if self._vllm_gate.enabled and x.dim() == 4 and not torch.compiler.is_compiling():
-            output = group_norm_silu_4d(
-                x,
-                self.weight,
-                self.bias,
-                self.num_groups,
-                self.eps,
-                apply_silu=True,
-            )
-            if output is not None:
-                return output
-        return F.silu(F.group_norm(x, self.num_groups, self.weight, self.bias, self.eps))
-
-
-def _install_norm_silu(decoder: nn.Module, resnet_cls: type[nn.Module], gate: VaeFastPathGate) -> int:
-    def is_fusable(norm: object) -> bool:
-        return type(norm) is nn.GroupNorm and norm.affine and norm.weight is not None and norm.bias is not None
-
-    count = 0
-    for module in decoder.modules():
-        if (
-            type(module) is resnet_cls
-            and module.time_emb_proj is None
-            and module.time_embedding_norm in ("default", "group")
-            and module.upsample is None
-            and module.downsample is None
-            and isinstance(module.nonlinearity, nn.SiLU)
-            and is_fusable(module.norm1)
-            and is_fusable(module.norm2)
-        ):
-            module.norm1 = FusedGroupNormSiLU(module.norm1, gate)
-            module.norm2 = FusedGroupNormSiLU(module.norm2, gate)
-            module.nonlinearity = nn.Identity()
-            count += 2
-
-    if (
-        type(getattr(decoder, "conv_norm_out", None)) is nn.GroupNorm
-        and isinstance(getattr(decoder, "conv_act", None), nn.SiLU)
-        and is_fusable(decoder.conv_norm_out)
-    ):
-        decoder.conv_norm_out = FusedGroupNormSiLU(decoder.conv_norm_out, gate)
-        decoder.conv_act = nn.Identity()
-        count += 1
-    return count
-
-
-_UPSAMPLE_TAP_MAP = {0: (2,), 1: (1, 2), 2: (0, 1), 3: (0,)}
-
-
-def _fold_upsample2x_conv2d_weight(conv: nn.Conv2d) -> torch.Tensor:
-    weight = conv.weight.detach().float()
-    output_channels, input_channels = weight.shape[:2]
-    folded = weight.new_zeros(input_channels, output_channels, 4, 4)
-    for row in range(4):
-        for column in range(4):
-            value = weight.new_zeros(output_channels, input_channels)
-            for source_row in _UPSAMPLE_TAP_MAP[row]:
-                for source_column in _UPSAMPLE_TAP_MAP[column]:
-                    value += weight[:, :, source_row, source_column]
-            folded[:, :, row, column] = value.t()
-    folded = folded.to(conv.weight.dtype)
-    if conv.weight.is_contiguous(memory_format=torch.channels_last):
-        folded = folded.contiguous(memory_format=torch.channels_last)
-    return folded.to(conv.weight.device)
-
-
-class FusedUpsample2xConv2d(nn.Module):
-    """Quality-gated nearest-2x + Conv2d replacement using ConvTranspose2d."""
-
-    def __init__(self, upsample: nn.Module, gate: VaeFastPathGate) -> None:
-        super().__init__()
-        # Keep only ``conv`` registered so checkpoint parameter names stay
-        # ``...upsamplers.N.conv.*``. The original module is an eager fallback.
-        object.__setattr__(self, "_original", upsample)
-        self.conv = upsample.conv
-        self.channels = upsample.channels
-        self._vllm_gate = gate
-        self.register_buffer("_fused_weight", None, persistent=False)
-
-    def forward(
-        self,
-        hidden_states: torch.Tensor,
-        output_size: int | None = None,
-        *args: Any,
-        **kwargs: Any,
-    ) -> torch.Tensor:
-        if not self._vllm_gate.enabled or output_size is not None or hidden_states.shape[1] != self.channels:
-            return self._original(hidden_states, output_size=output_size, *args, **kwargs)
-
-        weight = self._fused_weight
-        if weight is None or weight.device != self.conv.weight.device or weight.dtype != self.conv.weight.dtype:
-            weight = _fold_upsample2x_conv2d_weight(self.conv)
-            self._fused_weight = weight
-        return F.conv_transpose2d(hidden_states, weight, self.conv.bias, stride=2, padding=1)
-
-
-def _install_fused_upsample(decoder: nn.Module, upsample_cls: type[nn.Module], gate: VaeFastPathGate) -> int:
-    count = 0
-    for block in decoder.up_blocks:
-        upsamplers = getattr(block, "upsamplers", None)
-        if not upsamplers:
-            continue
-        for index, upsample in enumerate(upsamplers):
-            if type(upsample) is not upsample_cls:
-                continue
-            conv = getattr(upsample, "conv", None)
-            if (
-                upsample.use_conv
-                and not upsample.use_conv_transpose
-                and upsample.interpolate
-                and upsample.norm is None
-                and upsample.name == "conv"
-                and type(conv) is nn.Conv2d
-                and conv.kernel_size == (3, 3)
-                and conv.stride == (1, 1)
-                and conv.padding == (1, 1)
-                and conv.dilation == (1, 1)
-                and conv.groups == 1
-                and conv.padding_mode == "zeros"
-            ):
-                upsamplers[index] = FusedUpsample2xConv2d(upsample, gate)
-                count += 1
-    return count
-
-
-def _fold_attention_value_projection(module: nn.Module) -> tuple[torch.Tensor, torch.Tensor]:
-    value_weight = module.to_v.weight.detach().float()
-    value_bias = module.to_v.bias.detach().float()
-    output_weight = module.to_out[0].weight.detach().float()
-    output_bias = module.to_out[0].bias.detach().float()
-    dtype = module.to_v.weight.dtype
-    return (
-        (output_weight @ value_weight).to(dtype).contiguous(),
-        (output_weight @ value_bias + output_bias).to(dtype),
-    )
-
-
-def _attention_fast_forward(
-    self: nn.Module,
-    hidden_states: torch.Tensor,
-    encoder_hidden_states: torch.Tensor | None = None,
-    attention_mask: torch.Tensor | None = None,
-    temb: torch.Tensor | None = None,
-    **cross_attention_kwargs: Any,
-) -> torch.Tensor:
-    if (
-        not self._vllm_gate.enabled
-        or encoder_hidden_states is not None
-        or attention_mask is not None
-        or temb is not None
-        or hidden_states.ndim != 4
-    ):
-        return type(self).forward(
-            self,
-            hidden_states,
-            encoder_hidden_states=encoder_hidden_states,
-            attention_mask=attention_mask,
-            temb=temb,
-            **cross_attention_kwargs,
-        )
-
-    residual = hidden_states
-    batch_size, channels, height, width = hidden_states.shape
-    hidden_states = hidden_states.permute(0, 2, 3, 1).reshape(batch_size, height * width, channels)
-
-    if self.group_norm is not None:
-        norm = self.group_norm
-        normalized = None
-        if not torch.compiler.is_compiling():
-            normalized = group_norm_silu_rows(
-                hidden_states,
-                norm.weight,
-                norm.bias,
-                norm.num_groups,
-                norm.eps,
-                apply_silu=False,
-            )
-        hidden_states = normalized if normalized is not None else norm(hidden_states.transpose(1, 2)).transpose(1, 2)
-
-    query = self.to_q(hidden_states)
-    key = self.to_k(hidden_states)
-    folded_weight = self._vllm_folded_value_weight
-    folded_bias = self._vllm_folded_value_bias
-    if (
-        folded_weight is None
-        or folded_bias is None
-        or folded_weight.device != hidden_states.device
-        or folded_weight.dtype != self.to_v.weight.dtype
-    ):
-        folded_weight, folded_bias = _fold_attention_value_projection(self)
-        self._vllm_folded_value_weight = folded_weight
-        self._vllm_folded_value_bias = folded_bias
-    value = F.linear(hidden_states, folded_weight, folded_bias)
-
-    output = F.scaled_dot_product_attention(query.unsqueeze(1), key.unsqueeze(1), value.unsqueeze(1))
-    output = output.squeeze(1).to(query.dtype)
-    output = self.to_out[1](output)
-    output = output.reshape(batch_size, height, width, channels).permute(0, 3, 1, 2)
-    if self.residual_connection:
-        output = output + residual
-    return output / self.rescale_output_factor
-
-
-def _attention_fast_compatible(module: nn.Module, attention_cls: type, processor_cls: type) -> bool:
-    return (
-        type(module) is attention_cls
-        and isinstance(module.processor, processor_cls)
-        and module.heads == 1
-        and module.scale_qk
-        and module.spatial_norm is None
-        and not module.norm_cross
-        and getattr(module, "norm_q", None) is None
-        and getattr(module, "norm_k", None) is None
-        and getattr(module, "add_k_proj", None) is None
-        and type(module.to_q) is nn.Linear
-        and type(module.to_k) is nn.Linear
-        and type(module.to_v) is nn.Linear
-        and type(module.to_out[0]) is nn.Linear
-        and isinstance(module.to_out[1], nn.Dropout)
-        and module.to_v.bias is not None
-        and module.to_out[0].bias is not None
-    )
-
-
-def _decoder_layout_forward(self: nn.Module, *args: Any, **kwargs: Any) -> torch.Tensor:
-    use_channels_last = self._vllm_gate.enabled
-    if use_channels_last != self._vllm_channels_last:
-        memory_format = torch.channels_last if use_channels_last else torch.contiguous_format
-        self.to(memory_format=memory_format)
-        self._vllm_channels_last = use_channels_last
-        logger.debug(
-            "%s decoder switched to %s layout",
-            self._vllm_label,
-            "channels_last" if use_channels_last else "contiguous",
-        )
-    return type(self).forward(self, *args, **kwargs)
-
-
-def maybe_optimize_image_vae(vae: nn.Module) -> nn.Module:
-    """Install image VAE decoder fast paths when the structure is fully supported."""
-
-    if not _HAS_TRITON or getattr(vae, "_vllm_vae_fast_path_installed", False):
-        return vae
-
-    from diffusers.models.attention_processor import Attention, AttnProcessor2_0
-    from diffusers.models.autoencoders import AutoencoderKL
-    from diffusers.models.autoencoders.vae import Decoder
-    from diffusers.models.resnet import ResnetBlock2D
-    from diffusers.models.upsampling import Upsample2D
-
-    supported_types: tuple[type[nn.Module], ...] = (AutoencoderKL,)
-    try:
-        from diffusers.models.autoencoders.autoencoder_kl_flux2 import AutoencoderKLFlux2
-
-        supported_types += (AutoencoderKLFlux2,)
-    except ImportError:  # pragma: no cover - older supported Diffusers builds
-        pass
-
-    if not isinstance(vae, supported_types):
-        return vae
-    decoder = getattr(vae, "decoder", None)
-    if type(decoder) is not Decoder:
-        return vae
-
-    attention_modules = [
-        module for module in decoder.modules() if _attention_fast_compatible(module, Attention, AttnProcessor2_0)
-    ]
-    attention_count = sum(isinstance(module, Attention) for module in decoder.modules())
-    if len(attention_modules) != attention_count:
-        logger.info(
-            "%s has %d/%d attention blocks without a layout-safe rewrite; skipping VAE fast paths",
-            type(vae).__name__,
-            attention_count - len(attention_modules),
-            attention_count,
-        )
-        return vae
-
-    gate = VaeFastPathGate()
-    decoder._vllm_gate = gate
-    decoder._vllm_label = type(vae).__name__
-    decoder._vllm_channels_last = False
-    decoder.forward = MethodType(_decoder_layout_forward, decoder)
-    upsample_count = _install_fused_upsample(decoder, Upsample2D, gate)
-    for module in attention_modules:
-        module._vllm_gate = gate
-        module.register_buffer("_vllm_folded_value_weight", None, persistent=False)
-        module.register_buffer("_vllm_folded_value_bias", None, persistent=False)
-        module.forward = MethodType(_attention_fast_forward, module)
-    norm_count = _install_norm_silu(decoder, ResnetBlock2D, gate)
-    register_vae_fast_path_gate(vae, gate)
-    vae._vllm_vae_fast_path_installed = True
-    logger.info(
-        "%s installed quality-gated VAE decoder fast paths "
-        "(%d upsamplers, %d attention blocks, %d GroupNorm+SiLU fusions)",
-        type(vae).__name__,
-        upsample_count,
-        len(attention_modules),
-        norm_count,
-    )
-    return vae
-
-
-__all__ = [
-    "FusedGroupNormSiLU",
-    "FusedUpsample2xConv2d",
-    "maybe_optimize_image_vae",
-]

vllm_omni/diffusion/vae_optimizations/optimize.py

@@ -1,70 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Pipeline-level installation for optional VAE optimizations."""
-
-from __future__ import annotations
-
-from torch import nn
-from vllm.logger import init_logger
-
-from vllm_omni.diffusion.data import OmniDiffusionConfig
-from vllm_omni.diffusion.offloader.module_collector import ModuleDiscovery
-from vllm_omni.diffusion.vae_optimizations.image import maybe_optimize_image_vae
-from vllm_omni.diffusion.vae_optimizations.precision import configure_wan_decode_precision
-from vllm_omni.diffusion.vae_optimizations.wan import maybe_optimize_wan_vae
-from vllm_omni.platforms import current_omni_platform
-
-logger = init_logger(__name__)
-
-_FOLDED_CACHE_NAMES = (
-    "_fused_weight",
-    "_vllm_folded_value_weight",
-    "_vllm_folded_value_bias",
-)
-
-
-def optimize_pipeline_vaes(pipeline: nn.Module, od_config: OmniDiffusionConfig) -> None:
-    """Configure declared VAE components once after checkpoint loading."""
-
-    precision = getattr(od_config, "vae_decode_precision", None)
-    vaes = ModuleDiscovery.discover(pipeline).vaes
-    for vae in vaes:
-        # An explicit precision override is a serving contract, so failures on
-        # a recognized Wan VAE must abort startup instead of being hidden.
-        configure_wan_decode_precision(vae, precision)
-
-    if not current_omni_platform.is_cuda():
-        return
-
-    for vae in vaes:
-        for optimizer in (maybe_optimize_image_vae, maybe_optimize_wan_vae):
-            try:
-                optimizer(vae)
-            except Exception:
-                # A wrapper may already have installed its exact fallback
-                # before a later optional rewrite failed. With no registered
-                # enabled gate, those wrappers remain on the reference path.
-                logger.warning(
-                    "Failed to install optional VAE fast paths on %s; using the reference path",
-                    type(vae).__name__,
-                    exc_info=True,
-                )
-
-
-def clear_pipeline_vae_fast_path_caches(pipeline: nn.Module) -> None:
-    """Release lazily folded buffers before the worker enters sleep mode."""
-
-    cleared = 0
-    for vae in ModuleDiscovery.discover(pipeline).vaes:
-        if not getattr(vae, "_vllm_vae_fast_path_installed", False):
-            continue
-        for module in vae.modules():
-            for name in _FOLDED_CACHE_NAMES:
-                if name in module._buffers and module._buffers[name] is not None:
-                    setattr(module, name, None)
-                    cleared += 1
-    if cleared:
-        logger.debug("Released %d lazy VAE fast-path buffers before sleep", cleared)
-
-
-__all__ = ["clear_pipeline_vae_fast_path_caches", "optimize_pipeline_vaes"]

vllm_omni/diffusion/vae_optimizations/precision.py

@@ -1,91 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Decode-only precision configuration for structurally split Wan VAEs."""
-
-from __future__ import annotations
-
-from types import MethodType
-from typing import Any
-
-import torch
-from torch import nn
-from vllm.logger import init_logger
-
-logger = init_logger(__name__)
-
-PRECISION_TO_DTYPE = {
-    "fp32": torch.float32,
-    "fp16": torch.float16,
-    "bf16": torch.bfloat16,
-}
-
-
-def _decode_with_precision(
-    self: nn.Module,
-    latents: torch.Tensor,
-    *args: Any,
-    **kwargs: Any,
-) -> Any:
-    return type(self).decode(
-        self,
-        latents.to(dtype=self._vllm_decode_dtype),
-        *args,
-        **kwargs,
-    )
-
-
-def _encode_in_fp32(
-    self: nn.Module,
-    samples: torch.Tensor,
-    *args: Any,
-    **kwargs: Any,
-) -> Any:
-    return type(self).encode(
-        self,
-        samples.to(dtype=torch.float32),
-        *args,
-        **kwargs,
-    )
-
-
-def configure_wan_decode_precision(vae: nn.Module, precision: str | None) -> nn.Module:
-    """Keep Wan encode in FP32 while materializing only decode modules lower."""
-
-    if precision is None:
-        return vae
-
-    from diffusers.models.autoencoders import AutoencoderKLWan
-
-    if not isinstance(vae, AutoencoderKLWan):
-        return vae
-    try:
-        decode_dtype = PRECISION_TO_DTYPE[precision]
-    except KeyError as exc:
-        raise ValueError(
-            f"Unsupported vae_decode_precision={precision!r}; expected one of {sorted(PRECISION_TO_DTYPE)}"
-        ) from exc
-
-    encoder = getattr(vae, "encoder", None)
-    quant_conv = getattr(vae, "quant_conv", None)
-    decoder = getattr(vae, "decoder", None)
-    post_quant_conv = getattr(vae, "post_quant_conv", None)
-    if not all(isinstance(module, nn.Module) for module in (encoder, decoder, post_quant_conv)):
-        logger.info("Wan VAE lacks separate encoder/decoder modules; skipping decode-only precision")
-        return vae
-
-    encoder.to(dtype=torch.float32)
-    if isinstance(quant_conv, nn.Module):
-        quant_conv.to(dtype=torch.float32)
-    decoder.to(dtype=decode_dtype)
-    post_quant_conv.to(dtype=decode_dtype)
-    vae.encode = MethodType(_encode_in_fp32, vae)
-    vae._vllm_decode_dtype = decode_dtype
-    vae.decode = MethodType(_decode_with_precision, vae)
-    logger.info(
-        "Wan VAE configured with FP32 encode and %s decode",
-        precision,
-    )
-    return vae
-
-
-__all__ = ["PRECISION_TO_DTYPE", "configure_wan_decode_precision"]

vllm_omni/diffusion/vae_optimizations/triton_group_norm_silu.py

@@ -1,237 +0,0 @@
-# ruff: noqa: N803
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Channels-last two-pass GroupNorm(+SiLU) Triton kernels.
-
-The kernels keep the channel dimension innermost throughout a 2D VAE
-decoder. Statistics are accumulated in FP32, then a separate apply pass
-performs the affine transform and optional SiLU epilogue. Unsupported inputs
-return ``None`` so callers can fail closed to the reference implementation.
-"""
-
-from __future__ import annotations
-
-import torch
-import triton
-import triton.language as tl
-
-_SUPPORTED_DTYPES = {torch.float16, torch.bfloat16, torch.float32}
-_MAX_CHANNELS = 2048
-
-
-@triton.jit
-def _gn_partial_rows_kernel(
-    x_ptr,
-    psum_ptr,
-    psq_ptr,
-    rows,
-    rows_per_prog,
-    C: tl.constexpr,
-    BLOCK_R: tl.constexpr,
-):
-    chunk = tl.program_id(0).to(tl.int64)
-    batch = tl.program_id(1).to(tl.int64)
-    num_chunks = tl.num_programs(0)
-    row0 = chunk * rows_per_prog
-    cols = tl.arange(0, C)
-    acc_sum = tl.zeros((C,), tl.float32)
-    acc_sq = tl.zeros((C,), tl.float32)
-    x_base = x_ptr + batch * rows * C
-    for row_offset in range(0, rows_per_prog, BLOCK_R):
-        row_indices = row0 + row_offset + tl.arange(0, BLOCK_R)
-        mask = row_indices < rows
-        x = tl.load(
-            x_base + row_indices[:, None] * C + cols[None, :],
-            mask=mask[:, None],
-            other=0.0,
-        ).to(tl.float32)
-        acc_sum += tl.sum(x, 0)
-        acc_sq += tl.sum(x * x, 0)
-    output_offsets = (batch * num_chunks + chunk) * C + cols
-    tl.store(psum_ptr + output_offsets, acc_sum)
-    tl.store(psq_ptr + output_offsets, acc_sq)
-
-
-@triton.jit
-def _gn_finalize_kernel(
-    psum_ptr,
-    psq_ptr,
-    weight_ptr,
-    bias_ptr,
-    scale_shift_ptr,
-    num_chunks,
-    group_numel,
-    eps,
-    channels,
-    CPG: tl.constexpr,
-    BLOCK_K: tl.constexpr,
-):
-    group = tl.program_id(0).to(tl.int64)
-    batch = tl.program_id(1).to(tl.int64)
-    cols = group * CPG + tl.arange(0, CPG)
-    acc_sum = tl.zeros((), tl.float32)
-    acc_sq = tl.zeros((), tl.float32)
-    for chunk0 in range(0, num_chunks, BLOCK_K):
-        chunks = chunk0 + tl.arange(0, BLOCK_K)
-        mask = chunks < num_chunks
-        offsets = (batch * num_chunks + chunks)[:, None] * channels + cols[None, :]
-        acc_sum += tl.sum(tl.load(psum_ptr + offsets, mask=mask[:, None], other=0.0))
-        acc_sq += tl.sum(tl.load(psq_ptr + offsets, mask=mask[:, None], other=0.0))
-    mean = acc_sum / group_numel
-    variance = acc_sq / group_numel - mean * mean
-    variance = tl.maximum(variance, 0.0)
-    reciprocal_std = tl.rsqrt(variance + eps)
-    weight = tl.load(weight_ptr + cols).to(tl.float32)
-    bias = tl.load(bias_ptr + cols).to(tl.float32)
-    scale = weight * reciprocal_std
-    shift = bias - mean * scale
-    tl.store(scale_shift_ptr + batch * 2 * channels + cols, scale)
-    tl.store(scale_shift_ptr + (batch * 2 + 1) * channels + cols, shift)
-
-
-@triton.jit
-def _gn_apply_rows_kernel(
-    x_ptr,
-    scale_shift_ptr,
-    y_ptr,
-    rows,
-    C: tl.constexpr,
-    BLOCK_R: tl.constexpr,
-    SILU: tl.constexpr,
-):
-    program = tl.program_id(0).to(tl.int64)
-    batch = tl.program_id(1).to(tl.int64)
-    cols = tl.arange(0, C)
-    scale = tl.load(scale_shift_ptr + batch * 2 * C + cols)
-    shift = tl.load(scale_shift_ptr + (batch * 2 + 1) * C + cols)
-    row_indices = program * BLOCK_R + tl.arange(0, BLOCK_R)
-    mask = row_indices < rows
-    offsets = batch * rows * C + row_indices[:, None] * C + cols[None, :]
-    x = tl.load(x_ptr + offsets, mask=mask[:, None], other=0.0).to(tl.float32)
-    y = x * scale[None, :] + shift[None, :]
-    if SILU:
-        y = y * tl.sigmoid(y)
-    tl.store(y_ptr + offsets, y.to(y_ptr.dtype.element_ty), mask=mask[:, None])
-
-
-def _group_norm_silu_rows(
-    x: torch.Tensor,
-    weight: torch.Tensor,
-    bias: torch.Tensor,
-    num_groups: int,
-    eps: float,
-    apply_silu: bool,
-) -> torch.Tensor:
-    batch_size, rows, channels = x.shape
-    channels_per_group = channels // num_groups
-    block_rows = max(1, 8192 // channels)
-    rows_per_program = block_rows * 32
-    num_chunks = triton.cdiv(rows, rows_per_program)
-    partial_sum = torch.empty((batch_size, num_chunks, channels), device=x.device, dtype=torch.float32)
-    partial_sq = torch.empty_like(partial_sum)
-    with torch.cuda.device(x.device):
-        _gn_partial_rows_kernel[(num_chunks, batch_size)](
-            x,
-            partial_sum,
-            partial_sq,
-            rows,
-            rows_per_program,
-            C=channels,
-            BLOCK_R=block_rows,
-            num_warps=4,
-        )
-        scale_shift = torch.empty((batch_size, 2, channels), device=x.device, dtype=torch.float32)
-        block_chunks = max(1, min(4096 // channels_per_group, triton.next_power_of_2(num_chunks)))
-        _gn_finalize_kernel[(num_groups, batch_size)](
-            partial_sum,
-            partial_sq,
-            weight,
-            bias,
-            scale_shift,
-            num_chunks,
-            rows * channels_per_group,
-            eps,
-            channels,
-            CPG=channels_per_group,
-            BLOCK_K=block_chunks,
-            num_warps=4,
-        )
-        output = torch.empty_like(x)
-        _gn_apply_rows_kernel[(triton.cdiv(rows, block_rows), batch_size)](
-            x,
-            scale_shift,
-            output,
-            rows,
-            C=channels,
-            BLOCK_R=block_rows,
-            SILU=apply_silu,
-            num_warps=4,
-        )
-    return output
-
-
-def _is_supported(
-    x: torch.Tensor,
-    weight: torch.Tensor | None,
-    bias: torch.Tensor | None,
-    num_groups: int,
-) -> bool:
-    if not (x.is_cuda and x.numel() > 0 and not torch.is_grad_enabled()):
-        return False
-    if x.requires_grad or x.dtype not in _SUPPORTED_DTYPES:
-        return False
-    if weight is None or bias is None:
-        return False
-    channels = x.shape[1] if x.dim() == 4 else x.shape[-1]
-    if weight.shape != (channels,) or bias.shape != (channels,):
-        return False
-    if not (
-        weight.device == bias.device == x.device
-        and weight.dtype == bias.dtype == x.dtype
-        and weight.is_contiguous()
-        and bias.is_contiguous()
-    ):
-        return False
-    if num_groups < 1 or channels % num_groups != 0:
-        return False
-    return triton.next_power_of_2(channels) == channels and channels <= _MAX_CHANNELS
-
-
-def group_norm_silu_4d(
-    x: torch.Tensor,
-    weight: torch.Tensor,
-    bias: torch.Tensor,
-    num_groups: int,
-    eps: float,
-    apply_silu: bool = True,
-) -> torch.Tensor | None:
-    """Run GroupNorm(+SiLU) on a channels-last ``(N, C, H, W)`` tensor."""
-
-    if x.dim() != 4 or not _is_supported(x, weight, bias, num_groups):
-        return None
-    batch_size, channels, height, width = x.shape
-    if not (channels > 1 and (height > 1 or width > 1) and x.is_contiguous(memory_format=torch.channels_last)):
-        return None
-    rows = x.permute(0, 2, 3, 1).reshape(batch_size, height * width, channels)
-    output = _group_norm_silu_rows(rows, weight, bias, num_groups, eps, apply_silu)
-    return output.reshape(batch_size, height, width, channels).permute(0, 3, 1, 2)
-
-
-def group_norm_silu_rows(
-    x: torch.Tensor,
-    weight: torch.Tensor,
-    bias: torch.Tensor,
-    num_groups: int,
-    eps: float,
-    apply_silu: bool = True,
-) -> torch.Tensor | None:
-    """Run GroupNorm(+SiLU) over contiguous ``(N, rows, C)`` input."""
-
-    if x.dim() != 3 or not x.is_contiguous():
-        return None
-    if not _is_supported(x, weight, bias, num_groups):
-        return None
-    return _group_norm_silu_rows(x, weight, bias, num_groups, eps, apply_silu)
-
-
-__all__ = ["group_norm_silu_4d", "group_norm_silu_rows"]

vllm_omni/diffusion/vae_optimizations/triton_wan_rmsnorm_silu.py

@@ -1,160 +0,0 @@
-# ruff: noqa: N803
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Channels-last-3D Wan VAE RMSNorm(+SiLU) Triton kernel."""
-
-from __future__ import annotations
-
-import torch
-import triton
-import triton.language as tl
-
-_SUPPORTED_DTYPES = {torch.float16, torch.bfloat16, torch.float32}
-_MAX_CHANNELS = 1024
-
-
-@triton.jit
-def _wan_rmsnorm_silu_kernel(
-    x_ptr,
-    gamma_ptr,
-    bias_ptr,
-    output_ptr,
-    channels: tl.constexpr,
-    time_size,
-    height_size,
-    width_size,
-    x_stride_batch,
-    x_stride_channel,
-    x_stride_time,
-    x_stride_height,
-    x_stride_width,
-    output_stride_batch,
-    output_stride_channel,
-    output_stride_time,
-    output_stride_height,
-    output_stride_width,
-    rms_scale,
-    eps,
-    has_bias: tl.constexpr,
-    BLOCK_C: tl.constexpr,
-):
-    row = tl.program_id(0).to(tl.int64)
-    channel_offsets = tl.arange(0, BLOCK_C)
-    channel_mask = channel_offsets < channels
-
-    width = row % width_size
-    remainder = row // width_size
-    height = remainder % height_size
-    remainder = remainder // height_size
-    time = remainder % time_size
-    batch = remainder // time_size
-
-    x_base = batch * x_stride_batch + time * x_stride_time + height * x_stride_height + width * x_stride_width
-    output_base = (
-        batch * output_stride_batch
-        + time * output_stride_time
-        + height * output_stride_height
-        + width * output_stride_width
-    )
-
-    x = tl.load(
-        x_ptr + x_base + channel_offsets * x_stride_channel,
-        mask=channel_mask,
-        other=0.0,
-    ).to(tl.float32)
-    norm = tl.sqrt(tl.sum(x * x, axis=0))
-    inverse_norm = 1.0 / tl.maximum(norm, eps)
-
-    # Match the eager dtype boundaries: normalize and scale in activation
-    # dtype, affine in the promoted dtype, then SiLU in FP32.
-    y = (x * inverse_norm).to(x_ptr.dtype.element_ty)
-    gamma = tl.load(gamma_ptr + channel_offsets, mask=channel_mask, other=1.0)
-    y = (y * rms_scale).to(x_ptr.dtype.element_ty)
-    y = (y.to(tl.float32) * gamma.to(tl.float32)).to(output_ptr.dtype.element_ty)
-    if has_bias:
-        bias = tl.load(bias_ptr + channel_offsets, mask=channel_mask, other=0.0)
-        y = (y.to(tl.float32) + bias.to(tl.float32)).to(output_ptr.dtype.element_ty)
-    y = y.to(tl.float32)
-    y = y * tl.sigmoid(y)
-    tl.store(
-        output_ptr + output_base + channel_offsets * output_stride_channel,
-        y,
-        mask=channel_mask,
-    )
-
-
-def _affine_supported(x: torch.Tensor, affine: torch.Tensor) -> bool:
-    return (
-        affine.is_cuda
-        and affine.device == x.device
-        and (affine.dtype == x.dtype or affine.dtype == torch.float32)
-        and affine.numel() == x.shape[1]
-    )
-
-
-def can_use_wan_rmsnorm_silu(
-    x: torch.Tensor,
-    gamma: torch.Tensor,
-    bias: torch.Tensor | None,
-) -> bool:
-    """Return whether ``wan_rmsnorm_silu`` supports this exact input."""
-
-    return (
-        x.is_cuda
-        and not torch.is_grad_enabled()
-        and not x.requires_grad
-        and x.dtype in _SUPPORTED_DTYPES
-        and x.ndim == 5
-        and x.numel() > 0
-        and 0 < x.shape[1] <= _MAX_CHANNELS
-        and x.is_contiguous(memory_format=torch.channels_last_3d)
-        and _affine_supported(x, gamma)
-        and (bias is None or _affine_supported(x, bias))
-    )
-
-
-def wan_rmsnorm_silu(
-    x: torch.Tensor,
-    gamma: torch.Tensor,
-    bias: torch.Tensor | None = None,
-    rms_scale: float | None = None,
-    eps: float = 1e-12,
-) -> torch.Tensor | None:
-    """Fuse ``SiLU(F.normalize(x, dim=1) * scale * gamma + bias)``."""
-
-    if not can_use_wan_rmsnorm_silu(x, gamma, bias):
-        return None
-
-    batch_size, channels, time_size, height_size, width_size = x.shape
-    output_dtype = torch.promote_types(x.dtype, gamma.dtype)
-    output = torch.empty_strided(x.shape, x.stride(), device=x.device, dtype=output_dtype)
-    block_channels = triton.next_power_of_2(channels)
-    num_warps = 1 if block_channels <= 64 else 4 if block_channels <= 512 else 8
-    gamma = gamma.reshape(channels).contiguous()
-    has_bias = bias is not None
-    bias = gamma if bias is None else bias.reshape(channels).contiguous()
-    if rms_scale is None:
-        rms_scale = channels**0.5
-
-    with torch.cuda.device(x.device):
-        _wan_rmsnorm_silu_kernel[(batch_size * time_size * height_size * width_size,)](
-            x,
-            gamma,
-            bias,
-            output,
-            channels,
-            time_size,
-            height_size,
-            width_size,
-            *x.stride(),
-            *output.stride(),
-            float(rms_scale),
-            eps,
-            has_bias,
-            BLOCK_C=block_channels,
-            num_warps=num_warps,
-        )
-    return output
-
-
-__all__ = ["can_use_wan_rmsnorm_silu", "wan_rmsnorm_silu"]

vllm_omni/diffusion/vae_optimizations/wan.py

@@ -1,162 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-"""Quality-gated CUDA fast path for Diffusers Wan VAE decoders."""
-
-from __future__ import annotations
-
-from types import MethodType
-from typing import Any
-
-import torch
-import torch.nn.functional as F
-from torch import nn
-from vllm.logger import init_logger
-
-from vllm_omni.diffusion.vae_optimizations.gate import (
-    VaeFastPathGate,
-    register_vae_fast_path_gate,
-)
-
-logger = init_logger(__name__)
-
-try:
-    from vllm_omni.diffusion.vae_optimizations.triton_wan_rmsnorm_silu import (
-        wan_rmsnorm_silu,
-    )
-
-    _HAS_TRITON = True
-except ImportError:  # pragma: no cover - exercised on builds without Triton
-    _HAS_TRITON = False
-
-
-class FusedWanRMSNormSiLU(nn.Module):
-    """Quality-gated Wan RMSNorm + SiLU with an exact eager fallback."""
-
-    def __init__(self, norm: nn.Module, gate: VaeFastPathGate) -> None:
-        super().__init__()
-        self.gamma = norm.gamma
-        self.bias = norm.bias
-        self.scale = float(norm.scale)
-        self._vllm_gate = gate
-
-    def forward(self, x: torch.Tensor) -> torch.Tensor:
-        if self._vllm_gate.enabled and not torch.compiler.is_compiling():
-            bias = self.bias if isinstance(self.bias, torch.Tensor) else None
-            output = wan_rmsnorm_silu(x, self.gamma, bias, rms_scale=self.scale)
-            if output is not None:
-                return output
-
-        needs_fp32_normalize = x.dtype in (torch.float16, torch.bfloat16) or any(
-            token in str(x.dtype) for token in ("float4_", "float8_")
-        )
-        normalized = F.normalize(x.float() if needs_fp32_normalize else x, dim=1).to(x.dtype)
-        return F.silu(normalized * self.scale * self.gamma + self.bias)
-
-
-def _plain_silu(activation: object) -> bool:
-    return isinstance(activation, nn.SiLU) and not activation.inplace
-
-
-def _fusable_norm(norm: object, norm_cls: type) -> bool:
-    return (
-        type(norm) is norm_cls
-        and getattr(norm, "channel_first", False)
-        and isinstance(getattr(norm, "gamma", None), torch.Tensor)
-    )
-
-
-def _install_norm_silu(decoder: nn.Module, gate: VaeFastPathGate) -> int | None:
-    from diffusers.models.autoencoders.autoencoder_kl_wan import WanResidualBlock, WanRMS_norm
-
-    residual_blocks = [module for module in decoder.modules() if isinstance(module, WanResidualBlock)]
-    eligible = [
-        module
-        for module in residual_blocks
-        if type(module) is WanResidualBlock
-        and _plain_silu(module.nonlinearity)
-        and _fusable_norm(module.norm1, WanRMS_norm)
-        and _fusable_norm(module.norm2, WanRMS_norm)
-    ]
-    if len(eligible) != len(residual_blocks):
-        logger.info(
-            "Wan VAE has %d/%d non-standard residual blocks; skipping fast path",
-            len(residual_blocks) - len(eligible),
-            len(residual_blocks),
-        )
-        return None
-    if not (
-        _fusable_norm(getattr(decoder, "norm_out", None), WanRMS_norm)
-        and _plain_silu(getattr(decoder, "nonlinearity", None))
-    ):
-        logger.info("Wan VAE has a non-standard output head; skipping fast path")
-        return None
-
-    count = 0
-    for module in eligible:
-        module.norm1 = FusedWanRMSNormSiLU(module.norm1, gate)
-        module.norm2 = FusedWanRMSNormSiLU(module.norm2, gate)
-        module.nonlinearity = nn.Identity()
-        count += 2
-    decoder.norm_out = FusedWanRMSNormSiLU(decoder.norm_out, gate)
-    decoder.nonlinearity = nn.Identity()
-    return count + 1
-
-
-def _set_wan_decoder_memory_format(decoder: nn.Module, *, channels_last: bool) -> None:
-    for module in decoder.modules():
-        if isinstance(module, nn.Conv3d):
-            module.to(memory_format=torch.channels_last_3d if channels_last else torch.contiguous_format)
-        elif isinstance(module, (nn.Conv2d, nn.ConvTranspose2d)):
-            module.to(memory_format=torch.channels_last if channels_last else torch.contiguous_format)
-
-
-def _decoder_layout_forward(self: nn.Module, *args: Any, **kwargs: Any) -> torch.Tensor:
-    use_channels_last = self._vllm_gate.enabled
-    if use_channels_last != self._vllm_channels_last:
-        _set_wan_decoder_memory_format(self, channels_last=use_channels_last)
-        self._vllm_channels_last = use_channels_last
-        logger.debug(
-            "Wan VAE decoder switched to %s layout",
-            "channels_last_3d" if use_channels_last else "contiguous",
-        )
-    return type(self).forward(self, *args, **kwargs)
-
-
-def maybe_optimize_wan_vae(vae: nn.Module) -> nn.Module:
-    """Install the Wan decoder fast path when its full structure is supported."""
-
-    if not _HAS_TRITON or getattr(vae, "_vllm_vae_fast_path_installed", False):
-        return vae
-
-    from diffusers.models.autoencoders import AutoencoderKLWan
-    from diffusers.models.autoencoders.autoencoder_kl_wan import WanDecoder3d
-
-    if not isinstance(vae, AutoencoderKLWan):
-        return vae
-    decoder = getattr(vae, "decoder", None)
-    if type(decoder) is not WanDecoder3d:
-        return vae
-
-    distributed_executor = getattr(vae, "distributed_executor", None)
-    parallel_mode = getattr(distributed_executor, "parallel_mode", "tile")
-    if parallel_mode in ("auto", "spatial_shard_height", "spatial_shard_width"):
-        logger.info("Wan spatial-shard decode is configured; skipping quality-gated VAE fast path")
-        return vae
-
-    gate = VaeFastPathGate()
-    norm_count = _install_norm_silu(decoder, gate)
-    if norm_count is None:
-        return vae
-    decoder._vllm_gate = gate
-    decoder._vllm_channels_last = False
-    decoder.forward = MethodType(_decoder_layout_forward, decoder)
-    register_vae_fast_path_gate(vae, gate)
-    vae._vllm_vae_fast_path_installed = True
-    logger.info(
-        "Wan VAE installed quality-gated decoder fast path (%d RMSNorm+SiLU fusions)",
-        norm_count,
-    )
-    return vae
-
-
-__all__ = ["FusedWanRMSNormSiLU", "maybe_optimize_wan_vae"]

vllm_omni/diffusion/worker/diffusion_model_runner.py

@@ -30,6 +30,9 @@ from vllm_omni.diffusion.cache.prompt_embed_cache import (
 from vllm_omni.diffusion.cache.selector import get_cache_backend
 from vllm_omni.diffusion.compile import regionally_compile
 from vllm_omni.diffusion.data import DiffusionOutput, OmniDiffusionConfig
+from vllm_omni.diffusion.distributed.autoencoders.autoencoder_kl_wan import (
+    prepare_pipeline_wan_vae_data_movement,
+)
 from vllm_omni.diffusion.forward_context import set_forward_context
 from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
 from vllm_omni.diffusion.models.interface import (
@@ -43,10 +46,6 @@ from vllm_omni.diffusion.offloader import get_offload_backend
 from vllm_omni.diffusion.registry import _NO_CACHE_ACCELERATION
 from vllm_omni.diffusion.request import OmniDiffusionRequest
 from vllm_omni.diffusion.sched.interface import DiffusionSchedulerOutput, KVPrefetchJob
-from vllm_omni.diffusion.vae_optimizations import (
-    optimize_pipeline_vaes,
-    use_pipeline_vae_fast_path,
-)
 from vllm_omni.diffusion.worker.input_batch import InputBatch, scatter_latents
 from vllm_omni.diffusion.worker.request_batch import DiffusionRequestBatch
 from vllm_omni.diffusion.worker.utils import (
@@ -247,9 +246,7 @@ class DiffusionModelRunner(OmniConnectorModelRunnerMixin):
                     custom_pipeline_name=custom_pipeline_name,
                     device=self.device,
                 )
-                # Keep precision rematerialization in the tagged weights pool
+                prepare_pipeline_wan_vae_data_movement(self.pipeline)
-                # so sleep level 1 can offload every resulting parameter.
-                optimize_pipeline_vaes(self.pipeline, self.od_config)
         time_after_load = time.perf_counter()
 
         logger.info(
@@ -506,11 +503,6 @@ class DiffusionModelRunner(OmniConnectorModelRunnerMixin):
         if require_request_batch_support and not getattr(self.pipeline, "supports_request_batch", False):
             raise RuntimeError(f"{type(self.pipeline).__name__} does not support request-batch forward.")
 
-        quality = getattr(reqs[0].sampling_params, "quality", None)
-        if any(getattr(req.sampling_params, "quality", None) != quality for req in reqs[1:]):
-            raise RuntimeError("A diffusion worker batch cannot mix request quality levels.")
-        use_fast_vae = quality == "high"
-
         # Use no_grad() for HSDP compatibility, inference_mode() otherwise for
         # better perf. HSDP2's fully_shard pre-forward hooks need tensor version
         # counters, which inference tensors do not track.
@@ -535,8 +527,7 @@ class DiffusionModelRunner(OmniConnectorModelRunnerMixin):
 
             with set_forward_context(vllm_config=self.vllm_config, omni_diffusion_config=od_config):
                 with record_function(record_name):
-                    with use_pipeline_vae_fast_path(self.pipeline, use_fast_vae):
+                    raw_outputs = self.pipeline.forward(batch)
-                        raw_outputs = self.pipeline.forward(batch)
                     outputs = _normalize_pipeline_outputs(
                         raw_outputs,
                         expected_count=len(reqs),
@@ -787,11 +778,7 @@ class DiffusionModelRunner(OmniConnectorModelRunnerMixin):
 
                             if should_decode:
                                 clear_pipeline_stage_durations(self.pipeline)
-                                with use_pipeline_vae_fast_path(
+                                result = self.pipeline.post_decode(req)
-                                    self.pipeline,
-                                    getattr(req.sampling, "quality", None) == "high",
-                                ):
-                                    result = self.pipeline.post_decode(req)
                                 if result is not None:
                                     self._attach_stepwise_metrics(
                                         req,

vllm_omni/diffusion/worker/diffusion_worker.py

@@ -576,14 +576,6 @@ class DiffusionWorker:
 
         usage_before = allocator.get_current_usage()
 
-        if self.model_runner is not None and self.model_runner.pipeline is not None:
-            from vllm_omni.diffusion.vae_optimizations import clear_pipeline_vae_fast_path_caches
-
-            # Folded weights are allocated lazily on the request path rather
-            # than in the tagged weight pool. Drop them before sleep so they
-            # neither remain resident nor need a separate CPU backup.
-            clear_pipeline_vae_fast_path_caches(self.model_runner.pipeline)
-
         if level == 2 and self.model_runner is not None:
             if hasattr(self.model_runner, "graph_runners"):
                 self.model_runner.graph_runners.clear()

vllm_omni/engine/arg_utils.py

@@ -491,7 +491,6 @@ class OrchestratorArgs:
     step_execution: bool = False
     vae_use_slicing: bool = False
     vae_use_tiling: bool = False
-    vae_decode_precision: str | None = None
     enable_multithread_weight_load: bool = True
     num_weight_load_threads: int = 4
     enable_cpu_offload: bool = False

vllm_omni/engine/async_omni_engine.py

@@ -1026,7 +1026,6 @@ class AsyncOmniEngine:
             "request_batch_max_wait_ms": kwargs.get("request_batch_max_wait_ms", 0.0),
             "vae_use_slicing": kwargs.get("vae_use_slicing", False),
             "vae_use_tiling": kwargs.get("vae_use_tiling", False),
-            "vae_decode_precision": kwargs.get("vae_decode_precision", None),
             "cache_backend": cache_backend,
             "cache_config": cache_config,
             "enable_cache_dit_summary": kwargs.get("enable_cache_dit_summary", False),

vllm_omni/entrypoints/cli/serve.py

@@ -629,15 +629,6 @@ class OmniServeCommand(CLISubcommand):
             action="store_true",
             help="Enable VAE tiling for memory optimization (useful for mitigating OOM issues).",
         )
-        omni_config_group.add_argument(
-            "--vae-decode-precision",
-            choices=("fp32", "fp16", "bf16"),
-            default=None,
-            help=(
-                "Optional Wan decode-only VAE precision. The encoder remains FP32; "
-                "unset preserves the pipeline's existing precision behavior."
-            ),
-        )
 
         # Parallel weight loading (faster diffusion startup)
         omni_config_group.add_argument(