[Bugfix] Skip HWR store-size scans when no limit is configured

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

docs/design/feature/host_weight_runtime.md

@@ -67,26 +67,6 @@ views and mapped ranges. A separate transport still decides whether to use
 registered mmap, private pinned staging, synchronous copies, or asynchronous
 H2D transfer.
 
-## Payload validation at startup
-
-Eligible diffusion loaders accept `--host-weight-runtime-validation` (or the
-`host_weight_runtime_validation` offline/stage configuration field):
-
-- `manifest_and_metadata` is the existing default. It validates identity,
-  metadata, tensor structure, and sizes, but does not detect payload-only
-  corruption.
-- `full_checksum` reads and hashes every payload on warm acquisition before
-  restoring weights. This adds startup I/O/CPU work proportional to artifact
-  bytes; it is not a per-inference check.
-
-For example, add `--host-weight-runtime-validation full_checksum` to an enabled
-HWR deployment. Preferred mode falls back to canonical loading and can publish
-a replacement after detecting corruption; required mode fails startup. Readers
-can use different validation levels in the same domain without changing artifact
-identity. Neither level authenticates an untrusted writer or prevents mutation
-after validation. Checksumming does not imply automatic recovery of a running
-service.
-
 ## Resolution behavior
 
 The loader resolves the immutable canonical source and computes the exact

tests/config/test_omni_config.py

@@ -193,37 +193,6 @@ def test_resolve_execution_mode_rejects_unknown_execution_type():
         omni_config_module._resolve_execution_mode("unknown_execution_type")
 
 
-@pytest.mark.parametrize("validation", ["manifest_and_metadata", "full_checksum"])
-def test_hwr_validation_survives_config_projection(validation: str):
-    from vllm_omni.diffusion.data import OmniDiffusionConfig
-
-    override = StageDeployConfig(stage_id=0, host_weight_runtime_validation=validation)
-    config = _from_pipeline_key(
-        "dreamzero",
-        deploy_config_path="dreamzero_tp1_cfg2",
-        cli_overrides={"host_weight_runtime_validation": validation},
-    )
-    stage = config.stage_by_id(0)
-    assert isinstance(stage, VllmOmniDiffusionStageConfig)
-    assert stage.diffusion_config.host_weight_runtime_validation == validation
-    projection = omni_config_module._DiffusionConfigProjection.from_kwargs(
-        host_weight_runtime_validation=override.host_weight_runtime_validation
-    )
-    assert projection.host_weight_runtime_validation == validation
-    assert OmniDiffusionConfig(host_weight_runtime_validation=validation).host_weight_runtime_validation == validation
-    assert omni_config_module._DiffusionConfigProjection().host_weight_runtime_validation == "manifest_and_metadata"
-    assert OmniDiffusionConfig().host_weight_runtime_validation == "manifest_and_metadata"
-
-
-@pytest.mark.parametrize("validation", ["fs_verity", "unknown", None])
-def test_hwr_validation_rejects_unsupported_values(validation):
-    from vllm_omni.diffusion.data import OmniDiffusionConfig
-
-    for config_class in (OmniDiffusionConfig, omni_config_module._DiffusionConfigProjection):
-        with pytest.raises(ValueError, match="host_weight_runtime_validation"):
-            config_class(host_weight_runtime_validation=validation)
-
-
 def test_from_pipeline_config_preserves_current_pipeline_config_object():
     omni_config = _from_pipeline_key("minicpmo_4_5")
     pipeline = _resolve_pipeline_or_skip("minicpmo_4_5")

tests/diffusion/attention/test_ulysses_uaa.py

@@ -27,202 +27,6 @@ from vllm_omni.platforms import current_omni_platform
 pytestmark = [pytest.mark.core_model, pytest.mark.diffusion, pytest.mark.cpu]
 
 
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_uaa_equal_rank_contract_skips_length_gather(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    from vllm_omni.diffusion.attention.parallel import ulysses
-
-    class FakeGroup:
-        ulysses_group = object()
-        ulysses_world_size = 2
-        ulysses_rank = 0
-        ring_world_size = 1
-
-    observed = []
-
-    def fake_all_to_all(pg, tensor, **kwargs):
-        observed.append((kwargs["seq_lens"], kwargs["padded_head_cnt"]))
-        return tensor, tensor.shape[2]
-
-    monkeypatch.setattr(ulysses, "_ulysses_all_to_all_any_qkv", fake_all_to_all)
-    monkeypatch.setattr(ulysses, "get_ulysses_mode", lambda **kwargs: "advanced_uaa")
-    monkeypatch.setattr(
-        ulysses,
-        "_all_gather_int",
-        lambda *args, **kwargs: pytest.fail("length gather must be skipped"),
-    )
-    strategy = ulysses.UlyssesParallelAttention(
-        FakeGroup(),
-        scatter_idx=2,
-        gather_idx=1,
-        use_sync=False,
-    )
-
-    with set_forward_context():
-        get_forward_context()._sp_equal_pad_stack.append(True)
-        strategy.pre_attention(
-            torch.zeros(1, 3, 28, 4),
-            torch.zeros(1, 3, 7, 4),
-            torch.zeros(1, 3, 7, 4),
-            None,
-        )
-
-    assert observed == [([3, 3], 32), ([3, 3], 8), ([3, 3], 8)]
-
-
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_uaa_hybrid_equal_rank_contract_skips_ulysses_gather_keeps_ring_check(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Contract + ring: the fast path skips only the Ulysses gather.
-
-    The ring equality check is a separate collective on ring_group and must
-    still run, receiving s_global derived from the contract.
-    """
-    from vllm_omni.diffusion.attention.parallel import ulysses
-
-    class FakeGroup:
-        ulysses_group = object()
-        ring_group = object()
-        ulysses_world_size = 2
-        ulysses_rank = 0
-        ring_world_size = 2
-
-    sp_group = FakeGroup()
-    observed = []
-    ring_checks = []
-
-    def fake_all_to_all(pg, tensor, **kwargs):
-        observed.append((kwargs["seq_lens"], kwargs["padded_head_cnt"]))
-        return tensor, tensor.shape[2]
-
-    def fake_gather_int(pg, value, *, device):
-        if pg is sp_group.ulysses_group:
-            pytest.fail("Ulysses length gather must be skipped under the contract")
-        assert pg is sp_group.ring_group
-        ring_checks.append(int(value))
-        return [int(value), int(value)]
-
-    monkeypatch.setattr(ulysses, "_ulysses_all_to_all_any_qkv", fake_all_to_all)
-    monkeypatch.setattr(ulysses, "_all_gather_int", fake_gather_int)
-    monkeypatch.setattr(ulysses, "get_ulysses_mode", lambda **kwargs: "advanced_uaa")
-    strategy = ulysses.UlyssesParallelAttention(
-        sp_group,
-        scatter_idx=2,
-        gather_idx=1,
-        use_sync=False,
-    )
-
-    with set_forward_context():
-        get_forward_context()._sp_equal_pad_stack.append(True)
-        strategy.pre_attention(
-            torch.zeros(1, 3, 28, 4),
-            torch.zeros(1, 3, 7, 4),
-            torch.zeros(1, 3, 7, 4),
-            None,
-        )
-
-    # Ring check ran: s_global = ulysses_world_size * local = 2 * 3.
-    assert ring_checks == [6]
-    assert observed == [([3, 3], 32), ([3, 3], 8), ([3, 3], 8)]
-
-
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_uaa_without_contract_keeps_dynamic_length_gather(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """No auto_pad boundary means shards may differ, so the gather must stay."""
-    from vllm_omni.diffusion.attention.parallel import ulysses
-
-    class FakeGroup:
-        ulysses_group = object()
-        ulysses_world_size = 2
-        ulysses_rank = 0
-        ring_world_size = 1
-
-    observed = []
-    monkeypatch.setattr(
-        ulysses,
-        "_all_gather_int",
-        lambda *args, **kwargs: [3, 4],
-    )
-    monkeypatch.setattr(ulysses, "get_ulysses_mode", lambda **kwargs: "advanced_uaa")
-
-    def fake_all_to_all(pg, tensor, **kwargs):
-        observed.append(kwargs["seq_lens"])
-        return tensor, tensor.shape[2]
-
-    monkeypatch.setattr(ulysses, "_ulysses_all_to_all_any_qkv", fake_all_to_all)
-    strategy = ulysses.UlyssesParallelAttention(
-        FakeGroup(),
-        scatter_idx=2,
-        gather_idx=1,
-        use_sync=False,
-    )
-
-    with set_forward_context():
-        strategy.pre_attention(
-            torch.zeros(1, 3, 28, 4),
-            torch.zeros(1, 3, 7, 4),
-            torch.zeros(1, 3, 7, 4),
-            None,
-        )
-
-    assert observed == [[3, 4], [3, 4], [3, 4]]
-
-
-@pytest.mark.core_model
-@pytest.mark.cpu
-def test_uaa_mixed_boundaries_keep_the_dynamic_length_gather(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """One manual boundary inside an auto_pad region voids the contract."""
-    from vllm_omni.diffusion.attention.parallel import ulysses
-
-    class FakeGroup:
-        ulysses_group = object()
-        ulysses_world_size = 2
-        ulysses_rank = 0
-        ring_world_size = 1
-
-    monkeypatch.setattr(ulysses, "get_ulysses_mode", lambda **kwargs: "advanced_uaa")
-    gathered = []
-
-    def fake_gather(*args, **kwargs):
-        gathered.append(args)
-        return [3, 4]
-
-    monkeypatch.setattr(ulysses, "_all_gather_int", fake_gather)
-    monkeypatch.setattr(
-        ulysses,
-        "_ulysses_all_to_all_any_qkv",
-        lambda pg, tensor, **kwargs: (tensor, tensor.shape[2]),
-    )
-    strategy = ulysses.UlyssesParallelAttention(
-        FakeGroup(),
-        scatter_idx=2,
-        gather_idx=1,
-        use_sync=False,
-    )
-
-    with set_forward_context():
-        ctx = get_forward_context()
-        ctx._sp_equal_pad_stack.extend([True, False])
-        assert not ctx.sp_rank_local_seq_lens_equal
-        strategy.pre_attention(
-            torch.zeros(1, 3, 28, 4),
-            torch.zeros(1, 3, 7, 4),
-            torch.zeros(1, 3, 7, 4),
-            None,
-        )
-
-    assert len(gathered) == 1
-
-
 @pytest.mark.core_model
 @pytest.mark.cpu
 def test_uaa_gqa_head_padding_preserves_the_query_to_kv_ratio(
@@ -830,187 +634,6 @@ def test_ulysses_uaa_matches_baseline(
                 pass
 
 
-def _run_contract_skip_case(
-    local_rank: int,
-    world_size: int,
-    init_method: str,
-    input_file: str,
-    num_heads: int,
-    head_size: int,
-    num_kv_heads: int | None,
-) -> None:
-    """Per-rank worker: slow path vs contract fast path on real rank pairs.
-
-    Phase 1 leaves the equal-pad stack empty so the Ulysses length all-gather
-    really runs (counted); phase 2 pushes the split hook's auto_pad marker and
-    replaces _all_gather_int with a raising guard. With ring_degree=1 any call
-    to it is the Ulysses length gather, so the guard proves the skip, and the
-    per-rank output comparison proves the fast path is numerically equivalent
-    to the slow path it replaces.
-    """
-    from vllm_omni.diffusion.attention.parallel import ulysses as ulysses_mod
-
-    device = torch.device(f"{current_omni_platform.device_type}:{local_rank}")
-    current_omni_platform.set_device(device)
-
-    os.environ["DIFFUSION_ATTENTION_BACKEND"] = "TORCH_SDPA"
-
-    init_distributed_environment(world_size=world_size, rank=local_rank, distributed_init_method=init_method)
-    initialize_model_parallel(
-        data_parallel_size=1,
-        cfg_parallel_size=1,
-        sequence_parallel_size=world_size,
-        ulysses_degree=world_size,
-        ring_degree=1,
-        tensor_parallel_size=1,
-        pipeline_parallel_size=1,
-    )
-
-    parallel_config = DiffusionParallelConfig(
-        pipeline_parallel_size=1,
-        data_parallel_size=1,
-        tensor_parallel_size=1,
-        sequence_parallel_size=world_size,
-        ulysses_degree=world_size,
-        ring_degree=1,
-        cfg_parallel_size=1,
-        ulysses_mode="advanced_uaa",
-    )
-    od_config = OmniDiffusionConfig(model="test", dtype=torch.float32, parallel_config=parallel_config)
-
-    with set_forward_context(omni_diffusion_config=od_config), set_current_diffusion_config(od_config):
-        attn = Attention(
-            num_heads=num_heads,
-            head_size=head_size,
-            causal=False,
-            softmax_scale=1.0 / (head_size**0.5),
-            num_kv_heads=num_kv_heads,
-        ).to(device=device, dtype=torch.float32)
-
-        with np.load(input_file, allow_pickle=False) as payload:
-            q_full = torch.from_numpy(payload["q"]).to(device=device)
-            k_full = torch.from_numpy(payload["k"]).to(device=device)
-            v_full = torch.from_numpy(payload["v"]).to(device=device)
-
-        def make_shards() -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
-            # Fresh contiguous shards for each phase; seq_len is divisible by
-            # world_size, so every rank gets the same local length.
-            return (
-                torch.tensor_split(q_full, world_size, dim=1)[local_rank].contiguous(),
-                torch.tensor_split(k_full, world_size, dim=1)[local_rank].contiguous(),
-                torch.tensor_split(v_full, world_size, dim=1)[local_rank].contiguous(),
-            )
-
-        # The Attention layer only enables SP communication when
-        # ForwardContext.sp_active is True; in production the split hook sets
-        # this (and pushes the auto_pad equal-length marker onto
-        # _sp_equal_pad_stack). Here we drive it directly.
-        ctx = get_forward_context()
-        ctx._sp_shard_depth = 1
-
-        def make_sdp_ctx() -> object:
-            # Fresh instance per phase: the @contextmanager instance releases
-            # its state on first exit and cannot be re-entered.
-            return torch.backends.cuda.sdp_kernel(enable_flash=False, enable_mem_efficient=False, enable_math=True)
-
-        # Phase 1: contract absent -> the length all-gather must really run.
-        real_gather = ulysses_mod._all_gather_int
-        gather_calls: list[object] = []
-
-        def counting_gather(pg: object, value: int, *, device: torch.device) -> list[int]:
-            gather_calls.append(pg)
-            return real_gather(pg, value, device=device)
-
-        ulysses_mod._all_gather_int = counting_gather
-        try:
-            q, k, v = make_shards()
-            with make_sdp_ctx():
-                out_slow = attn(q, k, v, attn_metadata=None).contiguous()
-        finally:
-            ulysses_mod._all_gather_int = real_gather
-        assert len(gather_calls) >= 1, "slow path must call the Ulysses length all-gather"
-
-        # Phase 2: replicate the split hook's auto_pad marker so the fast
-        # path fires; any _all_gather_int call in this phase is the Ulysses
-        # length gather and must not happen.
-        ctx._sp_equal_pad_stack.append(True)
-
-        def fail_on_gather(pg: object, value: int, *, device: torch.device) -> list[int]:
-            raise AssertionError("the Ulysses length all-gather must be skipped under the equal-pad contract")
-
-        ulysses_mod._all_gather_int = fail_on_gather
-        try:
-            q, k, v = make_shards()
-            with make_sdp_ctx():
-                out_fast = attn(q, k, v, attn_metadata=None).contiguous()
-        finally:
-            ulysses_mod._all_gather_int = real_gather
-        ctx._sp_equal_pad_stack.pop()
-
-        torch.testing.assert_close(out_fast, out_slow, atol=1e-5, rtol=1e-5)
-
-    destroy_distributed_env()
-
-
-@pytest.mark.parametrize(
-    "num_heads,num_kv_heads",
-    [
-        (3, None),  # MHA: heads padded 3 -> 4 by ulysses_degree=2
-        (28, 7),  # GQA: derived-Q padding 28/7 -> 32/8, the main case for the skip
-    ],
-)
-def test_ulysses_uaa_contract_skips_length_gather_matches_slow_path(num_heads: int, num_kv_heads: int | None) -> None:
-    """Real 2-rank parity: contract fast path == slow path, gather skipped.
-
-    The CPU-only contract tests replace the all-to-all with a fake; this test
-    runs the real strategy on two ranks. Each rank first computes the slow
-    path (empty equal-pad stack, real length all-gather), then the fast path
-    with the split hook's auto_pad marker pushed and _all_gather_int replaced
-    by a raising guard -- which also proves the collective was truly skipped,
-    not just silently replaced.
-    """
-    sp_world_size = 2
-    if current_omni_platform.get_device_count() < sp_world_size:
-        pytest.skip(f"Test requires {sp_world_size} GPUs")
-
-    batch_size = 2
-    head_size = 8
-    seq_len = 6  # divisible by 2 so both ranks hold equal-length shards
-
-    init_method = get_distributed_init_method()
-
-    with tempfile.NamedTemporaryFile(delete=False, suffix=".npz") as f_in:
-        input_file = f_in.name
-
-    try:
-        torch.manual_seed(0)
-        kv_heads = num_kv_heads if num_kv_heads is not None else num_heads
-        np.savez(
-            input_file,
-            q=torch.randn(batch_size, seq_len, num_heads, head_size, dtype=torch.float32).numpy(),
-            k=torch.randn(batch_size, seq_len, kv_heads, head_size, dtype=torch.float32).numpy(),
-            v=torch.randn(batch_size, seq_len, kv_heads, head_size, dtype=torch.float32).numpy(),
-        )
-
-        torch.multiprocessing.spawn(
-            _run_contract_skip_case,
-            args=(
-                sp_world_size,
-                init_method,
-                input_file,
-                num_heads,
-                head_size,
-                num_kv_heads,
-            ),
-            nprocs=sp_world_size,
-        )
-    finally:
-        try:
-            os.remove(input_file)
-        except OSError:
-            pass
-
-
 @pytest.mark.parametrize(
     "num_heads,num_kv_heads,joint_len",
     [

tests/diffusion/distributed/test_sp_plan_hooks.py

@@ -1113,120 +1113,6 @@ class TestStrictModeSplitValidation:
                 hook._prepare_sp_input(x, metadata["x"], (), {})
 
 
-@pytest.mark.cpu
-class TestEqualPadContract:
-    """The stack of per-boundary auto_pad flags that lets attention skip a gather.
-
-    A boundary is recorded as equal-length only when the framework padded it,
-    because only then is every rank's shard guaranteed the same size.
-    """
-
-    def _split_hook(self, *, auto_pad: bool, monkeypatch):
-        from vllm_omni.diffusion.attention import selector
-        from vllm_omni.diffusion.distributed import parallel_state, sp_sharding
-        from vllm_omni.diffusion.distributed.sp_plan import SequenceParallelConfig
-        from vllm_omni.diffusion.hooks.sequence_parallel import (
-            SequenceParallelSplitHook,
-        )
-
-        class MaskBackend:
-            supports_attention_mask = True
-
-            @staticmethod
-            def get_name():
-                return "test"
-
-        monkeypatch.setattr(parallel_state, "get_sequence_parallel_world_size", lambda: 2)
-        monkeypatch.setattr(parallel_state, "get_sequence_parallel_rank", lambda: 1)
-        monkeypatch.setattr(parallel_state, "get_ring_parallel_world_size", lambda: 1)
-        monkeypatch.setattr(selector, "get_attn_backend_for_role", lambda **_: (MaskBackend(), None))
-        # sp_shard, used by the manual path, binds these at import time.
-        monkeypatch.setattr(sp_sharding, "get_sequence_parallel_world_size", lambda: 2)
-        monkeypatch.setattr(sp_sharding, "get_sequence_parallel_rank", lambda: 1)
-
-        hook = SequenceParallelSplitHook(
-            {
-                0: SequenceParallelInput(
-                    split_dim=1,
-                    expected_dims=3,
-                    split_output=True,
-                    auto_pad=auto_pad,
-                )
-            },
-            SequenceParallelConfig(ulysses_degree=2),
-        )
-        hook.initialize_hook(nn.Identity())
-        return hook
-
-    def test_auto_pad_split_claims_equal_rank_lengths(self, monkeypatch):
-        from vllm_omni.diffusion.forward_context import (
-            get_forward_context,
-            set_forward_context,
-        )
-
-        hook = self._split_hook(auto_pad=True, monkeypatch=monkeypatch)
-        with set_forward_context():
-            hook.pre_forward(nn.Identity())
-            shard = hook.post_forward(nn.Identity(), torch.arange(5).reshape(1, 5, 1))
-            ctx = get_forward_context()
-
-            # 5 tokens padded to 6, so both ranks hold 3.
-            assert shard.shape[1] == 3
-            assert ctx._sp_equal_pad_stack == [True]
-            assert ctx.sp_rank_local_seq_lens_equal
-
-    def test_manual_split_does_not_claim_equal_rank_lengths(self, monkeypatch):
-        from vllm_omni.diffusion.forward_context import (
-            get_forward_context,
-            set_forward_context,
-        )
-
-        hook = self._split_hook(auto_pad=False, monkeypatch=monkeypatch)
-        with set_forward_context():
-            hook.pre_forward(nn.Identity())
-            hook.post_forward(nn.Identity(), torch.arange(6).reshape(1, 6, 1))
-            ctx = get_forward_context()
-
-            assert ctx._sp_equal_pad_stack == [False]
-            assert not ctx.sp_rank_local_seq_lens_equal
-
-    def test_gather_pops_only_its_own_boundary(self, monkeypatch):
-        """A stack, not a counter: the gather must not resurrect the contract.
-
-        With one auto_pad and one manual boundary open, a counter could not
-        tell which one a gather closes and would wrongly report equal lengths.
-        """
-        from vllm_omni.diffusion.distributed.sp_plan import SequenceParallelConfig
-        from vllm_omni.diffusion.forward_context import (
-            get_forward_context,
-            set_forward_context,
-        )
-        from vllm_omni.diffusion.hooks import sequence_parallel
-        from vllm_omni.diffusion.hooks.sequence_parallel import (
-            SequenceParallelGatherHook,
-        )
-
-        monkeypatch.setattr(
-            sequence_parallel,
-            "sp_gather",
-            lambda tensor, dim, validate: torch.cat([tensor, tensor], dim=dim),
-        )
-        hook = SequenceParallelGatherHook(
-            SequenceParallelOutput(gather_dim=1, expected_dims=3),
-            SequenceParallelConfig(ulysses_degree=2),
-        )
-
-        with set_forward_context():
-            ctx = get_forward_context()
-            ctx._sp_shard_depth = 2
-            ctx._sp_equal_pad_stack.extend([False, True])
-            hook.post_forward(nn.Identity(), torch.zeros(1, 3, 4))
-
-            assert ctx._sp_shard_depth == 1
-            assert ctx._sp_equal_pad_stack == [False]
-            assert not ctx.sp_rank_local_seq_lens_equal
-
-
 @pytest.mark.cpu
 class TestSequenceParallelConfig:
     """Test SequenceParallelConfig dataclass."""

tests/diffusion/model_loader/test_diffusers_loader.py

@@ -260,78 +260,6 @@ def test_hwr_cold_publication_and_warm_restore_skip_ordinary_dit_loading(
     assert len(tuple((store_root / "source-digests-v1" / "entries").glob("*.json"))) == 1
 
 
-@pytest.mark.parametrize("validation", ["manifest_and_metadata", "full_checksum"])
-@pytest.mark.parametrize("mode", ["preferred", "required"])
-def test_hwr_loader_payload_integrity_policy(tmp_path: Path, monkeypatch, validation: str, mode: str):
-    from vllm_omni.diffusion.offloader.startup import take_offload_startup_state
-
-    canonical = tmp_path / "canonical"
-    canonical.mkdir()
-    expected = torch.arange(4, dtype=torch.float32).to(torch.bfloat16).reshape(2, 2)
-    save_file({"weight": expected}, str(canonical / "model.safetensors"))
-    root = tmp_path / "store"
-
-    def make_loader(runtime_mode: str):
-        config = _hwr_config(canonical, root, mode=runtime_mode)
-        config.host_weight_runtime_validation = validation
-        loader = DiffusersPipelineLoader(LoadConfig(), config)
-        pipeline = _HWRPipeline(canonical)
-        monkeypatch.setattr(loader, "_init_from_load_format", lambda *args, **kwargs: pipeline)
-        return loader, pipeline
-
-    def close_plan(pipeline):
-        state = take_offload_startup_state(pipeline)
-        if state is not None and state.host_weight_plan is not None:
-            carrier = state.host_weight_plan.lease_carrier
-            if carrier is not None:
-                carrier.close()
-
-    cold_loader, cold = make_loader("preferred")
-    cold_loader.load_model(load_device="cpu", device=torch.device("cpu"))
-    assert cold.load_count == 1
-    close_plan(cold)
-    payloads = list((root / "artifacts").glob("*/*.safetensors"))
-    assert len(payloads) == 1
-    payload = payloads[0]
-    data = bytearray(payload.read_bytes())
-    data[-1] ^= 1  # Keep the header and file length valid; change only a BF16 payload byte.
-    original_mode = payload.stat().st_mode
-    payload.chmod(0o600)
-    payload.write_bytes(data)
-    payload.chmod(original_mode)
-
-    loader, pipeline = make_loader(mode)
-    try:
-        if validation == "full_checksum" and mode == "required":
-            with pytest.raises(RuntimeError, match="Host Weight Runtime resolution failed"):
-                loader.load_model(load_device="cpu", device=torch.device("cpu"))
-            assert pipeline.load_count == 0
-            assert loader.take_host_weight_plan() is None
-        else:
-            loader.load_model(load_device="cpu", device=torch.device("cpu"))
-            if validation == "manifest_and_metadata":
-                assert pipeline.load_count == 0
-                assert not torch.equal(pipeline.transformer.weight, expected)
-            else:
-                assert pipeline.load_count == 1
-                assert torch.equal(pipeline.transformer.weight, expected)
-    finally:
-        close_plan(pipeline)
-
-    if validation == "full_checksum":
-        if mode == "required":
-            assert list((root / "deny").iterdir())
-        else:
-            assert list((root / "quarantine").iterdir())
-            warm_loader, warm = make_loader(mode)
-            try:
-                warm_loader.load_model(load_device="cpu", device=torch.device("cpu"))
-                assert warm.load_count == 0
-                assert torch.equal(warm.transformer.weight, expected)
-            finally:
-                close_plan(warm)
-
-
 def test_maybe_fuse_distilled_lora_skips_when_hwr_warm_snapshot_present():
     cfg = SimpleNamespace(
         lora_backend="distill",

tests/entrypoints/test_async_omni_diffusion_config.py

@@ -503,8 +503,6 @@ def test_serve_cli_forwards_hwr_policy_for_no_allgather_dlo():
             "--omni",
             "--enable-distributed-layerwise-offload",
             "--dlo-no-use-allgather",
-            "--host-weight-runtime-validation",
-            "full_checksum",
             "--host-weight-runtime-mode",
             "preferred",
             "--host-weight-runtime-root",
@@ -521,8 +519,6 @@ def test_serve_cli_forwards_hwr_policy_for_no_allgather_dlo():
     assert explicit_kwargs["host_weight_runtime_mode"] == "preferred"
     assert explicit_kwargs["host_weight_runtime_root"] == "/var/cache/vllm-omni/hwr"
     assert explicit_kwargs["dlo_host_registration_limit_gib"] == 80
-    assert explicit_kwargs["host_weight_runtime_validation"] == "full_checksum"
-    assert engine_args["host_weight_runtime_validation"] == "full_checksum"
     assert engine_args["host_weight_runtime_mode"] == "preferred"
     assert engine_args["host_weight_runtime_root"] == "/var/cache/vllm-omni/hwr"
     assert engine_args["dlo_host_registration_limit_gib"] == 80

vllm_omni/config/omni_config.py

@@ -739,7 +739,6 @@ class _DiffusionConfigProjection:
     dlo_resident_layers: int = Field(default=0, ge=0)
     host_weight_runtime_mode: Literal["disabled", "preferred", "required"] = "disabled"
     host_weight_runtime_root: str | None = None
-    host_weight_runtime_validation: str = "manifest_and_metadata"
     dlo_host_registration_limit_gib: float = Field(default=0.0, ge=0)
     pin_cpu_memory: bool = True
     diffusion_compile_granularity: Literal["regional", "full"] = "regional"
@@ -893,7 +892,6 @@ class _DiffusionConfigProjection:
         validate_host_weight_runtime_options(
             mode=self.host_weight_runtime_mode,
             root=self.host_weight_runtime_root,
-            validation=self.host_weight_runtime_validation,
         )
         self.dlo_host_registration_limit_gib = validate_dlo_host_registration_options(
             limit_gib=self.dlo_host_registration_limit_gib,

vllm_omni/config/stage_config.py

@@ -483,7 +483,6 @@ class StageDeployConfig:
     dlo_resident_layers: int | None = None
     host_weight_runtime_mode: str | None = None
     host_weight_runtime_root: str | None = None
-    host_weight_runtime_validation: str | None = None
     dlo_host_registration_limit_gib: float | None = None
     # Diffusion-specific debug and observability knobs.
     enable_diffusion_pipeline_profiler: bool | None = None

vllm_omni/diffusion/attention/parallel/ulysses.py

@@ -14,11 +14,7 @@ from vllm_omni.diffusion.attention.backends.abstract import AttentionMetadata
 from vllm_omni.diffusion.attention.parallel.base import ParallelAttentionContext
 from vllm_omni.diffusion.distributed.comm import SeqAllToAll4D
 from vllm_omni.diffusion.distributed.group_coordinator import SequenceParallelGroupCoordinator
-from vllm_omni.diffusion.forward_context import (
-    get_forward_context,
-    get_ulysses_mode,
-    is_forward_context_available,
-)
+from vllm_omni.diffusion.forward_context import get_ulysses_mode
 
 logger = init_logger(__name__)
 
@@ -146,7 +142,7 @@ def _ulysses_all_to_all_any_qkv(
 
     input_split_sizes = [s_local] * world_size
     output_split_sizes = seq_lens
-    s_global = sum(output_split_sizes)
+    s_global = int(sum(output_split_sizes))
 
     out = torch.empty((s_global, bsz, head_cnt_local, head_dim), device=x.device, dtype=x.dtype)
     dist.all_to_all_single(
@@ -181,7 +177,7 @@ def _ulysses_all_to_all_any_o(
         return x
 
     bsz, s_global, head_cnt_local, head_dim = x.shape
-    s_local = local_seq_len
+    s_local = int(local_seq_len)
 
     # (B, S_global, H_local, D) -> (S_global, B, H_local, D)
     x_t = x.permute(1, 0, 2, 3).contiguous()
@@ -415,22 +411,9 @@ class UlyssesParallelAttention:
                     f"(got scatter_idx={self._scatter_idx}, gather_idx={self._gather_idx})."
                 )
 
-            if is_forward_context_available() and get_forward_context().sp_rank_local_seq_lens_equal:
-                # auto_pad already made every rank's shard the same length, so
-                # the lengths are known without a collective. Keep local_seq_len
-                # as a SymInt (no int()) so torch.compile(dynamic=True) does not
-                # specialize on it, and skip the host sync + graph break that
-                # _all_gather_int would introduce in every attention call.
-                local_seq_len = query.shape[1]
-                seq_lens = [local_seq_len] * ulysses_world_size
-            else:
-                local_seq_len = int(query.shape[1])
-                seq_lens = _all_gather_int(
-                    self._ulysses_pg,
-                    local_seq_len,
-                    device=query.device,
-                )
-            s_global = sum(seq_lens)
+            local_seq_len = int(query.shape[1])
+            seq_lens = _all_gather_int(self._ulysses_pg, local_seq_len, device=query.device)
+            s_global = int(sum(seq_lens))
 
             # In hybrid Ulysses+Ring, Ring attention uses P2P send/recv with fixed-shape
             # buffers. This requires all ring ranks to have the same seq_len after the
@@ -588,8 +571,8 @@ class UlyssesParallelAttention:
             joint_len=joint_len,
             joint_strategy=joint_strategy,
             use_uaa=(mode == "advanced_uaa"),
-            uaa_seq_lens=tuple(seq_lens) if mode == "advanced_uaa" else (),
-            uaa_local_seq_len=local_seq_len if mode == "advanced_uaa" else 0,
+            uaa_seq_lens=tuple(int(x) for x in seq_lens) if mode == "advanced_uaa" else (),
+            uaa_local_seq_len=int(local_seq_len) if mode == "advanced_uaa" else 0,
             orig_head_cnt=int(orig_head_cnt) if mode == "advanced_uaa" else 0,
             joint_orig_head_cnt=int(joint_orig_head_cnt) if mode == "advanced_uaa" else 0,
         )

vllm_omni/diffusion/data.py

@@ -100,9 +100,7 @@ def normalize_omni_diffusion_kwargs(kwargs: Mapping[str, Any]) -> dict[str, Any]
     return normalized
 
 
-def validate_host_weight_runtime_options(
-    *, mode: object, root: object, validation: object = "manifest_and_metadata"
-) -> None:
+def validate_host_weight_runtime_options(*, mode: object, root: object) -> None:
     """Validate HWR policy without touching the configured storage domain.
 
     Filesystem locality and store construction belong to the eligible loader
@@ -111,8 +109,6 @@ def validate_host_weight_runtime_options(
     """
     if mode not in {"disabled", "preferred", "required"}:
         raise ValueError("host_weight_runtime_mode must be disabled, preferred, or required")
-    if validation not in ("manifest_and_metadata", "full_checksum"):
-        raise ValueError("host_weight_runtime_validation must be manifest_and_metadata or full_checksum")
     if mode != "disabled" and (not isinstance(root, str) or not root.strip()):
         raise ValueError("enabled Host Weight Runtime requires host_weight_runtime_root")
 
@@ -835,7 +831,6 @@ class OmniDiffusionConfig:
     # existing loader/storage path.
     host_weight_runtime_mode: str = "disabled"
     host_weight_runtime_root: str | None = None
-    host_weight_runtime_validation: str = "manifest_and_metadata"
     # Optional per-worker ceiling for registering final-layout HWR mappings.
     # Zero adds no ceiling; pin_cpu_memory controls whether registration is tried.
     dlo_host_registration_limit_gib: float = 0.0
@@ -1246,7 +1241,6 @@ class OmniDiffusionConfig:
         validate_host_weight_runtime_options(
             mode=self.host_weight_runtime_mode,
             root=self.host_weight_runtime_root,
-            validation=self.host_weight_runtime_validation,
         )
         self.dlo_host_registration_limit_gib = validate_dlo_host_registration_options(
             limit_gib=self.dlo_host_registration_limit_gib,

vllm_omni/diffusion/forward_context.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 from contextlib import contextmanager
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
 import torch
@@ -66,10 +66,6 @@ class ForwardContext:
     # Tracks the depth of SP sharding - incremented on shard, decremented on gather
     # Used by attention layers to determine if SP communication should be enabled
     _sp_shard_depth: int = 0
-    # One entry per active SP split boundary: whether it auto-pads, which makes
-    # every rank's shard the same length and lets attention collectives skip a
-    # runtime length all-gather. Pushed on split, popped on gather.
-    _sp_equal_pad_stack: list[bool] = field(default_factory=list)
 
     @property
     def sp_active(self) -> bool:
@@ -92,15 +88,6 @@ class ForwardContext:
         sp_size = self.omni_diffusion_config.parallel_config.sequence_parallel_size
         return sp_size is not None and sp_size > 1
 
-    @property
-    def sp_rank_local_seq_lens_equal(self) -> bool:
-        """Whether every active SP boundary guarantees equal local shard sizes.
-
-        Only framework-managed auto_pad boundaries provide this contract; a
-        region that also shards manually keeps the dynamic length exchange.
-        """
-        return bool(self._sp_equal_pad_stack) and all(self._sp_equal_pad_stack)
-
     def __post_init__(self):
         pass
 

vllm_omni/diffusion/hooks/sequence_parallel.py

@@ -264,7 +264,6 @@ class SequenceParallelSplitHook(ModelHook):
 
         output_list = [output] if is_tensor else list(output)
         actually_sharded = False
-        equal_rank_seq_lens = True
 
         for index, spm in self.metadata.items():
             if not isinstance(index, int):
@@ -278,13 +277,10 @@ class SequenceParallelSplitHook(ModelHook):
             output_list[index] = self._prepare_sp_input(original, spm, self._last_args, self._last_kwargs)
             if output_list[index] is not original:
                 actually_sharded = True
-                equal_rank_seq_lens &= isinstance(spm, SequenceParallelInput) and spm.auto_pad
 
         # Mark SP as active only if at least one tensor was actually sharded
         if actually_sharded and is_forward_context_available():
-            ctx = get_forward_context()
-            ctx._sp_shard_depth += 1
-            ctx._sp_equal_pad_stack.append(equal_rank_seq_lens)
+            get_forward_context()._sp_shard_depth += 1
 
         return output_list[0] if is_tensor else type(output)(output_list)
 
@@ -557,8 +553,6 @@ class SequenceParallelGatherHook(ModelHook):
         if actually_gathered and is_forward_context_available():
             ctx = get_forward_context()
             ctx._sp_shard_depth = max(0, ctx._sp_shard_depth - 1)
-            if ctx._sp_equal_pad_stack:
-                ctx._sp_equal_pad_stack.pop()
 
         return output[0] if is_tensor else type(output)(output)
 

vllm_omni/diffusion/model_loader/host_weight_loader.py

@@ -26,7 +26,6 @@ from vllm_omni.diffusion.model_loader.host_weight_plan import HostWeightPlan, Te
 logger = init_logger(__name__)
 
 if TYPE_CHECKING:
-    from vllm_omni.diffusion.data import OmniDiffusionConfig
     from vllm_omni.diffusion.model_loader.host_weights.identity_adapter import FinalLayoutIdentityContext
     from vllm_omni.diffusion.model_loader.host_weights.source_identity import (
         NodeSourceDigestCache,
@@ -41,8 +40,6 @@ class _HWRCommitError(RuntimeError):
 class HWRLoaderMixin:
     """Optional final-layout HWR behavior shared by diffusion loaders."""
 
-    od_config: OmniDiffusionConfig
-
     @staticmethod
     def _identity_value(value: object) -> object:
         """Convert config objects into deterministic identity metadata."""
@@ -322,12 +319,10 @@ class HWRLoaderMixin:
             HostWeightLeaseCarrier,
             HostWeightRuntime,
             HostWeightRuntimeConfig,
-            IntegrityPolicy,
             ProductionPolicy,
             ResolutionOutcome,
             RuntimeMode,
             StorageDomainPolicy,
-            ValidationLevel,
         )
         from vllm_omni.host_weight_runtime.filesystem import detect_storage_class
 
@@ -368,11 +363,6 @@ class HWRLoaderMixin:
             HostWeightRuntimeConfig(
                 mode=mode,
                 domain=StorageDomainPolicy(root=root, storage_class=detect_storage_class(root)),
-                integrity=IntegrityPolicy(
-                    local_lookup=ValidationLevel(
-                        getattr(self.od_config, "host_weight_runtime_validation", "manifest_and_metadata")
-                    ),
-                ),
                 production=ProductionPolicy(
                     allow_local_build=False,
                     allow_post_load_publish=True,

vllm_omni/engine/arg_utils.py

@@ -582,7 +582,6 @@ class OrchestratorArgs:
     dlo_resident_layers: int = 0
     host_weight_runtime_mode: str = "disabled"
     host_weight_runtime_root: str | None = None
-    host_weight_runtime_validation: str = "manifest_and_metadata"
     dlo_host_registration_limit_gib: float = 0.0
     boundary_ratio: float | None = None
     flow_shift: float | None = None

vllm_omni/engine/async_omni_engine.py

@@ -1145,7 +1145,6 @@ class AsyncOmniEngine:
             "dlo_resident_layers": kwargs.get("dlo_resident_layers", 0),
             "host_weight_runtime_mode": kwargs.get("host_weight_runtime_mode", "disabled"),
             "host_weight_runtime_root": kwargs.get("host_weight_runtime_root"),
-            "host_weight_runtime_validation": kwargs.get("host_weight_runtime_validation", "manifest_and_metadata"),
             "dlo_host_registration_limit_gib": kwargs.get("dlo_host_registration_limit_gib", 0.0),
             "enforce_eager": False if kwargs.get("enforce_eager") is None else kwargs.get("enforce_eager"),
             "diffusion_compile_granularity": (

vllm_omni/entrypoints/cli/serve.py

@@ -770,15 +770,6 @@ class OmniServeCommand(CLISubcommand):
                 "preferred first."
             ),
         )
-        omni_config_group.add_argument(
-            "--host-weight-runtime-validation",
-            choices=("manifest_and_metadata", "full_checksum"),
-            default="manifest_and_metadata",
-            help=(
-                "HWR warm-acquisition validation. The default checks metadata, not payload corruption; "
-                "full_checksum reads and hashes every payload before restoring weights."
-            ),
-        )
         omni_config_group.add_argument(
             "--host-weight-runtime-root",
             type=str,

vllm_omni/host_weight_runtime/filesystem/store.py

@@ -1247,8 +1247,10 @@ class FilesystemHostWeightStore:
                     retryable=True,
                 )
             )
-        store_bytes = self._tree_bytes(self.root)
-        if policy.max_store_bytes is not None and store_bytes + additional_bytes > policy.max_store_bytes:
+        if (
+            policy.max_store_bytes is not None
+            and self._tree_bytes(self.root) + additional_bytes > policy.max_store_bytes
+        ):
             raise HostWeightError(
                 _failure(
                     ResolutionStage.CAPACITY,