[Core] Add opt-in serialized HWR build admission

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
@@ -333,6 +313,16 @@ One process per exact identity owns a build; other workers wait and then acquire
 leases for the published artifact. Publication is invisible until all payloads
 and metadata are validated, hashed, fsynced, and atomically renamed.
 
+By default, different identities may build concurrently. A domain configured
+with `CapacityPolicy(build_admission="serialized", max_store_bytes=...)` admits
+only one producer/publication operation at a time. Warm hits remain concurrent,
+and admission uses the existing coordination deadline. All workers must agree
+on the serialized domain's schema-2 capacity policy; use a new root when opting
+in, since in-place migration of an active domain is unsupported. This prevents
+competing producers from racing allocation checks but does not provide a
+filesystem-wide byte quota or automatic stale-data reclamation. See the
+[module capacity contract](../module/host_weight_runtime.md#validation-and-capacity).
+
 `coordination_timeout_seconds` bounds filesystem lock acquisition. It does not
 cancel synchronous validation, a producer that has already started, or atomic
 publication. A hung in-process producer therefore blocks its owning process and

docs/design/module/host_weight_runtime.md

@@ -311,9 +311,41 @@ point-in-time evidence only; kernel locks are not a persistent owner registry.
 Capacity policy covers all store-owned bytes, including ready, temporary, and
 quarantined data. The local writer preflights `max_artifact_bytes`,
 `max_store_bytes`, and `min_free_bytes`, and preallocates payloads where the
-filesystem supports it. `ENOSPC` remains a normal typed store failure because
-concurrent preflight is inherently racy. Automatic eviction and strict
-concurrent reservations are not implemented.
+filesystem supports it. `ENOSPC` remains a normal typed store failure.
+
+`CapacityPolicy.build_admission` defaults to `concurrent`, preserving parallel
+production for different identities and best-effort capacity preflight. An
+explicit `serialized` policy, which requires `max_store_bytes`, admits one
+cooperative build/publication operation per domain through
+`locks/domain-build.lock`. Production lock order becomes domain admission,
+per-key build, then artifact. Validated warm hits bypass admission; queued
+same-key callers still recheck and join a completed artifact. Admission waiting
+uses the caller's existing coordination deadline and returns a retryable build
+timeout without interrupting its owner.
+
+The kernel releases admission on process exit. An explicit same-key retry uses
+the existing stale-temp recovery; files from unrelated dead builds continue to
+count toward capacity. A synchronous producer that hangs still requires
+external process supervision. Admission release errors are logged without
+revising an already determined publication result.
+
+Concurrent domains retain their exact schema-1 capacity document. Serialized
+domains use a schema-2 capacity document with `build_admission=serialized`, so
+mixed policies and old readers fail initialization rather than bypass admission.
+Opt into serialization using a new root with an agreed policy; in-place policy
+migration while old workers may exist is unsupported. For example:
+
+```python
+from vllm_omni.host_weight_runtime import CapacityPolicy
+
+capacity = CapacityPolicy(max_store_bytes=192 * 1024**3, build_admission="serialized")
+```
+
+Serialization prevents cooperative producers from racing each other's
+allocation checks. It is not a filesystem quota: control metadata, finalization,
+and outside writers can still affect total usage. Automatic eviction and
+parallel byte reservations are not implemented. This policy trades parallel
+cold-build throughput for predictable producer admission.
 
 For tmpfs, artifact bytes are host-memory consumption and may consume swap;
 they must not be reported operationally as ordinary disk capacity. The store

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/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

tests/host_weight_runtime/test_build_admission.py

@@ -0,0 +1,305 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
+"""CPU process and policy coverage for domain-wide producer admission."""
+
+from __future__ import annotations
+
+import errno
+import json
+import multiprocessing as mp
+import os
+import time
+from multiprocessing.connection import Connection
+from pathlib import Path
+from typing import Literal
+
+import pytest
+import torch
+
+from tests.host_weight_runtime.test_filesystem_store import FakeProducer, _identity, _make_store, _publish_test_artifact
+from vllm_omni.host_weight_runtime import (
+    BuildRequest,
+    CapacityPolicy,
+    FailureCode,
+    HostWeightError,
+    HostWeightRuntime,
+    HostWeightRuntimeConfig,
+    ProductionMetadata,
+    ResolutionOutcome,
+    RuntimeMode,
+    StoreResult,
+    StoreStatus,
+    TensorWriteSpec,
+    ValidationLevel,
+    WaitPolicy,
+    WeightArtifactIdentity,
+    WeightProductionSpec,
+)
+from vllm_omni.host_weight_runtime.filesystem.locks import FileLock, lock_is_active
+from vllm_omni.host_weight_runtime.protocols import ArtifactWriter
+
+pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
+
+
+class AdmissionProducer:
+    def __init__(self, identity: WeightArtifactIdentity, control: Connection | None = None) -> None:
+        self._spec = FakeProducer(identity).spec
+        self.control = control
+
+    @property
+    def spec(self) -> WeightProductionSpec:
+        return self._spec
+
+    def produce(self, writer: ArtifactWriter) -> ProductionMetadata:
+        action = "produce"
+        if self.control is not None:
+            self.control.send("entered")
+            if not self.control.poll(60):
+                raise TimeoutError("test did not release producer")
+            action = self.control.recv()
+        spec = TensorWriteSpec("payload", (65536,), torch.bfloat16)
+        with writer.open_tensor_file("weights.safetensors", (spec,)) as output:
+            output.write_tensor("payload", torch.arange(65536, dtype=torch.float32).to(torch.bfloat16))
+            if action == "crash":
+                os._exit(27)
+        return ProductionMetadata("test-producer-v1", "test-restorer-v1")
+
+
+def _admission_process(root: Path, policy: CapacityPolicy, rank: int, control: Connection) -> None:
+    try:
+        store = _make_store(root, capacity=policy)
+        identity = _identity(tp_rank=rank)
+        original_lookup = store.lookup
+        first_lookup = True
+
+        def notify_lookup(
+            identity: WeightArtifactIdentity, *, validation: ValidationLevel, deadline: float | None = None
+        ) -> StoreResult:
+            nonlocal first_lookup
+            result = original_lookup(identity, validation=validation, deadline=deadline)
+            if first_lookup:
+                # Prove the caller observed a miss before the parent lets the
+                # active producer publish, independent of process scheduling.
+                control.send("looked_up")
+                first_lookup = False
+            return result
+
+        store.lookup = notify_lookup  # type: ignore[method-assign]
+        control.send("ready")
+        assert control.poll(60) and control.recv() == "start"
+        result = store.get_or_build(
+            BuildRequest(identity),
+            AdmissionProducer(identity, control),
+            validation=ValidationLevel.FULL_CHECKSUM,
+            deadline=time.monotonic() + 30,
+        )
+        if result.lease is not None:
+            result.lease.close()
+        control.send((result.status.value, result.failure.code.value if result.failure is not None else None))
+    finally:
+        control.close()
+
+
+@pytest.mark.parametrize("admission", ["concurrent", "serialized"])
+@pytest.mark.parametrize("same_key", [False, True])
+def test_cross_identity_admission_and_capacity(
+    tmp_path: Path, admission: Literal["concurrent", "serialized"], same_key: bool
+) -> None:
+    policy = CapacityPolicy(max_store_bytes=192 * 1024, build_admission=admission)
+    store = _make_store(tmp_path / "store", capacity=policy)
+    ctx = mp.get_context("spawn")
+    pairs = [ctx.Pipe() for _ in range(2)]
+    processes = [
+        ctx.Process(target=_admission_process, args=(store.root, policy, 0 if same_key else rank, pair[1]))
+        for rank, pair in enumerate(pairs)
+    ]
+    try:
+        for process, pair in zip(processes, pairs, strict=True):
+            process.start()
+            pair[1].close()
+        first, second = [pair[0] for pair in pairs]
+        for connection in (first, second):
+            assert connection.poll(60) and connection.recv() == "ready"
+        first.send("start")
+        assert first.poll(10) and first.recv() == "looked_up"
+        assert first.poll(10) and first.recv() == "entered"
+        second.send("start")
+        assert second.poll(10) and second.recv() == "looked_up"
+        if admission == "concurrent" and not same_key:
+            assert second.poll(10) and second.recv() == "entered"
+        else:
+            assert not second.poll(0.2), "a second producer entered while coordination should exclude it"
+        first.send("produce")
+        assert first.poll(30) and first.recv() == (StoreStatus.BUILT.value, None)
+        if same_key:
+            assert second.poll(30) and second.recv() == (StoreStatus.JOINED.value, None)
+        else:
+            if admission == "serialized":
+                assert second.poll(10) and second.recv() == "entered"
+            second.send("produce")
+            assert second.poll(30)
+            assert second.recv() == (StoreStatus.FAILED.value, FailureCode.STORE_LIMIT_EXCEEDED.value)
+        assert policy.max_store_bytes is not None
+        assert store.inspect_domain().store_bytes < policy.max_store_bytes
+        assert not list(store.tmp_dir.iterdir())
+        assert {path.name for path in store.artifacts_dir.iterdir()} == {_identity(tp_rank=0).key}
+        for process in processes:
+            process.join(10)
+            assert process.exitcode == 0
+    finally:
+        for process in processes:
+            if process.pid is not None:
+                if process.is_alive():
+                    process.kill()
+                process.join(5)
+                process.close()
+        for pair in pairs:
+            for connection in pair:
+                connection.close()
+
+
+def test_admission_timeout_preserves_owner_and_warm_hit_bypasses_lock(tmp_path: Path) -> None:
+    policy = CapacityPolicy(max_store_bytes=512 * 1024, build_admission="serialized")
+    store = _make_store(tmp_path / "store", capacity=policy)
+    identity, _ = _publish_test_artifact(store)
+    other = _identity(tp_rank=1)
+    lock_path = store.locks_dir / "domain-build.lock"
+    with FileLock(lock_path, exclusive=True, deadline=None):
+        hit = store.get_or_build(
+            BuildRequest(identity),
+            FakeProducer(identity),
+            validation=ValidationLevel.FULL_CHECKSUM,
+            deadline=time.monotonic() + 0.02,
+        )
+        assert hit.status is StoreStatus.HIT and hit.lease is not None
+        hit.lease.close()
+        timeout = store.get_or_build(
+            BuildRequest(other),
+            FakeProducer(other),
+            validation=ValidationLevel.FULL_CHECKSUM,
+            deadline=time.monotonic() + 0.02,
+        )
+        assert timeout.status is StoreStatus.TIMEOUT
+        assert timeout.failure is not None and timeout.failure.code is FailureCode.ACTIVE_BUILD_TIMEOUT
+        assert timeout.failure.retryable
+        assert lock_is_active(lock_path)
+    _publish_test_artifact(store, other)
+
+
+def test_producer_failure_releases_admission(tmp_path: Path) -> None:
+    store = _make_store(
+        tmp_path / "store", capacity=CapacityPolicy(max_store_bytes=16384, build_admission="serialized")
+    )
+    identity = _identity()
+    failed = store.get_or_build(
+        BuildRequest(identity),
+        FakeProducer(identity, write_mode="incomplete"),
+        validation=ValidationLevel.FULL_CHECKSUM,
+        deadline=time.monotonic() + 5,
+    )
+    assert failed.status is StoreStatus.FAILED
+    assert not lock_is_active(store.locks_dir / "domain-build.lock")
+    _publish_test_artifact(store, identity)
+
+
+@pytest.mark.parametrize(
+    ("mode", "expected"),
+    [(RuntimeMode.PREFERRED, ResolutionOutcome.CANONICAL_FALLBACK), (RuntimeMode.REQUIRED, ResolutionOutcome.FAILED)],
+)
+def test_admission_timeout_obeys_runtime_mode(tmp_path: Path, mode: RuntimeMode, expected: ResolutionOutcome) -> None:
+    policy = CapacityPolicy(max_store_bytes=16384, build_admission="serialized")
+    store = _make_store(tmp_path / "store", capacity=policy)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(mode=mode, domain=store.domain_policy, capacity=policy, wait=WaitPolicy(0.01))
+    )
+    identity = _identity()
+    with FileLock(store.locks_dir / "domain-build.lock", exclusive=True, deadline=None):
+        resolution = runtime.resolve(identity, producer=FakeProducer(identity))
+    assert resolution.report.outcome is expected
+    failure = resolution.report.attempts[-1].failure
+    assert failure is not None and failure.code is FailureCode.ACTIVE_BUILD_TIMEOUT
+
+
+def test_crashed_producer_releases_admission_and_same_key_retry_recovers(tmp_path: Path) -> None:
+    policy = CapacityPolicy(max_store_bytes=192 * 1024, build_admission="serialized")
+    store = _make_store(tmp_path / "store", capacity=policy)
+    ctx = mp.get_context("spawn")
+    parent, child = ctx.Pipe()
+    process = ctx.Process(target=_admission_process, args=(store.root, policy, 0, child))
+    process.start()
+    child.close()
+    try:
+        assert parent.poll(60) and parent.recv() == "ready"
+        parent.send("start")
+        assert parent.poll(10) and parent.recv() == "looked_up"
+        assert parent.poll(10) and parent.recv() == "entered"
+        parent.send("crash")
+        process.join(10)
+        assert process.exitcode == 27
+    finally:
+        if process.is_alive():
+            process.kill()
+        process.join(5)
+        process.close()
+        parent.close()
+    assert not lock_is_active(store.locks_dir / "domain-build.lock")
+    assert list(store.tmp_dir.iterdir())
+    identity = _identity(tp_rank=0)
+    recovered = store.get_or_build(
+        BuildRequest(identity),
+        AdmissionProducer(identity),
+        validation=ValidationLevel.FULL_CHECKSUM,
+        deadline=time.monotonic() + 10,
+    )
+    assert recovered.status is StoreStatus.BUILT and recovered.lease is not None
+    assert torch.equal(recovered.lease.tensors["payload"], torch.arange(65536, dtype=torch.float32).to(torch.bfloat16))
+    recovered.lease.close()
+    assert not list(store.tmp_dir.iterdir())
+
+
+def test_admission_release_error_preserves_publication(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    store = _make_store(
+        tmp_path / "store", capacity=CapacityPolicy(max_store_bytes=16384, build_admission="serialized")
+    )
+    original_close = FileLock.close
+
+    def fail_admission_close(lock: FileLock) -> None:
+        original_close(lock)
+        if lock.path.name == "domain-build.lock":
+            raise OSError(errno.EIO, "injected admission release failure")
+
+    monkeypatch.setattr(FileLock, "close", fail_admission_close)
+    _publish_test_artifact(store)
+    assert not lock_is_active(store.locks_dir / "domain-build.lock")
+
+
+def test_admission_policy_is_authoritative_and_legacy_document_is_unchanged(tmp_path: Path) -> None:
+    concurrent = CapacityPolicy(max_store_bytes=16384)
+    serialized = CapacityPolicy(max_store_bytes=16384, build_admission="serialized")
+    legacy = _make_store(tmp_path / "legacy", capacity=concurrent)
+    path = legacy.root / "domain-policy.json"
+    original = path.read_bytes()
+    assert json.loads(original) == {
+        "schema_version": 1,
+        "policy_version": 1,
+        "max_artifact_bytes": None,
+        "max_store_bytes": 16384,
+        "min_free_bytes": 0,
+        "eviction": "none",
+    }
+    with pytest.raises(HostWeightError, match="incompatible schema"):
+        _make_store(legacy.root, capacity=serialized)
+    assert path.read_bytes() == original
+    strict = _make_store(tmp_path / "serialized", capacity=serialized)
+    document = json.loads((strict.root / "domain-policy.json").read_bytes())
+    assert document["schema_version"] == 2 and document["build_admission"] == "serialized"
+    assert _make_store(strict.root, capacity=serialized).domain_uuid == strict.domain_uuid
+    with pytest.raises(HostWeightError, match="incompatible schema"):
+        _make_store(strict.root, capacity=concurrent)
+
+
+def test_admission_policy_validation() -> None:
+    with pytest.raises(ValueError, match="requires max_store_bytes"):
+        CapacityPolicy(build_admission="serialized")
+    with pytest.raises(ValueError, match="must be concurrent or serialized"):
+        CapacityPolicy(build_admission="unknown")  # type: ignore[arg-type]

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/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/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/config.py

@@ -8,6 +8,7 @@ import math
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
+from typing import Literal
 
 
 class RuntimeMode(str, Enum):
@@ -66,6 +67,7 @@ class CapacityPolicy:
     max_store_bytes: int | None = None
     min_free_bytes: int = 0
     eviction: str = "none"
+    build_admission: Literal["concurrent", "serialized"] = "concurrent"
 
     def __post_init__(self) -> None:
         for name, value in (
@@ -78,6 +80,10 @@ class CapacityPolicy:
             raise ValueError("min_free_bytes must not be negative")
         if not isinstance(self.eviction, str) or self.eviction != "none":
             raise ValueError("automatic host weight eviction is not implemented")
+        if self.build_admission not in ("concurrent", "serialized"):
+            raise ValueError("build_admission must be concurrent or serialized")
+        if self.build_admission == "serialized" and self.max_store_bytes is None:
+            raise ValueError("serialized build admission requires max_store_bytes")
 
 
 @dataclass(frozen=True)

vllm_omni/host_weight_runtime/filesystem/store.py

@@ -627,7 +627,7 @@ class FilesystemHostWeightStore:
                 self.policy_version = 1
 
     def _capacity_document(self, *, policy_version: int) -> dict[str, object]:
-        return {
+        document: dict[str, object] = {
             "schema_version": 1,
             "policy_version": policy_version,
             "max_artifact_bytes": self.capacity_policy.max_artifact_bytes,
@@ -635,6 +635,9 @@ class FilesystemHostWeightStore:
             "min_free_bytes": self.capacity_policy.min_free_bytes,
             "eviction": self.capacity_policy.eviction,
         }
+        if self.capacity_policy.build_admission == "serialized":
+            document.update(schema_version=2, build_admission="serialized")
+        return document
 
     def _build_lock_path(self, key: str) -> Path:
         return self.locks_dir / f"{key}.build.lock"
@@ -1013,6 +1016,40 @@ class FilesystemHostWeightStore:
         if initial.status in {StoreStatus.TIMEOUT, StoreStatus.FAILED}:
             return initial
 
+        if self.capacity_policy.build_admission == "concurrent":
+            return self._build_after_miss(identity, producer, validation=validation, deadline=deadline)
+        try:
+            admission = FileLock(self.locks_dir / "domain-build.lock", exclusive=True, deadline=deadline)
+        except FileLockTimeoutError as exc:
+            return StoreResult(
+                StoreStatus.TIMEOUT,
+                failure=_failure(
+                    ResolutionStage.PRODUCTION, FailureCode.ACTIVE_BUILD_TIMEOUT, str(exc), retryable=True
+                ),
+            )
+        except OSError as exc:
+            return StoreResult(
+                StoreStatus.FAILED,
+                failure=_failure(ResolutionStage.DOMAIN, FailureCode.DOMAIN_UNAVAILABLE, str(exc), retryable=True),
+            )
+        try:
+            return self._build_after_miss(identity, producer, validation=validation, deadline=deadline)
+        finally:
+            try:
+                admission.close()
+            except OSError:
+                # Releasing coordination must not revise the completed build's
+                # publication outcome or replace its primary failure.
+                logger.warning("Failed to release domain build admission under %s", self.root, exc_info=True)
+
+    def _build_after_miss(
+        self,
+        identity: WeightArtifactIdentity,
+        producer: WeightProducer,
+        *,
+        validation: ValidationLevel,
+        deadline: float,
+    ) -> StoreResult:
         try:
             build_lock = FileLock(self._build_lock_path(identity.key), exclusive=True, deadline=deadline)
         except FileLockTimeoutError as exc: