[Core] Make post-load publication warm-only

Close publication leases inside Host Weight Runtime, return only the publication report, and cover the new terminal mappings. Rename the runtime test module to avoid the pytest basename collision reported in #6436.

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

docs/design/feature/host_weight_runtime.md

@@ -24,6 +24,7 @@ V1 includes:
 - coordinated local lookup and one-producer publication;
 - descriptor-backed safetensors mmap leases;
 - preferred and required resolution policy;
+- explicit, separately reported post-load publication;
 - validation, deny, quarantine, cleanup, and capacity controls; and
 - typed reports for every terminal resolution outcome.
 
@@ -33,7 +34,6 @@ V1 does not include:
 - a generic BF16, FP8, quantized, or model-specific producer;
 - CUDA registration, pinned staging, H2D scheduling, or GPU kernels;
 - a remote artifact provider or cross-node coordination;
-- post-load publication (V1 accepts only pre-load-safe producers);
 - automatic eviction; or
 - a change to DLO AllGather or no-AllGather behavior.
 
@@ -84,6 +84,12 @@ flowchart TD
     F -->|"preferred"| C
     F -->|"required"| E["Fail startup"]
     P -->|"nonretryable failure"| E
+    C --> W{"Post-load publication enabled?"}
+    W -->|"yes"| B2["Publish final model through POST_LOAD_ONLY producer"]
+    B2 --> CL["Close validated publication lease"]
+    CL --> R2["Return separate publication report"]
+    R2 --> D
+    W -->|"no"| D["Keep canonical model"]
 ```
 
 The modes are:
@@ -94,10 +100,12 @@ The modes are:
 | `preferred` | Return an exact lease when available; otherwise use canonical fallback only for a miss, invalid cache entry, or typed retryable failure. |
 | `required` | Fail when an exact lease cannot be acquired. |
 
-An unsupported capability, semantic identity collision, producer failure, or
-publication failure is nonretryable and remains visible even in preferred
-mode. Storage policy must never disguise a semantic or configuration error as
-a cache miss.
+During pre-load resolution, an unsupported capability, semantic identity
+collision, producer failure, or publication failure is nonretryable and remains
+visible even in preferred mode. A post-load publication failure is likewise
+visible in its own report, but cannot revise the canonical-fallback outcome.
+Storage policy must never disguise a semantic or configuration error as a cache
+miss.
 
 ## Loader and restoration sequence
 
@@ -118,24 +126,35 @@ sequenceDiagram
     R->>S: lookup(exact identity)
     alt validated hit
         S-->>R: HostWeightLease
+        R-->>L: lease and LOCAL_HIT report
+        L->>X: plan_restore(model, lease)
+        X-->>L: validation-only restore plan
+        L->>X: commit() once
     else miss with allowed producer
         R->>S: get_or_build(identity, producer)
         S->>P: produce(store-scoped writer)
         P-->>S: final-layout tensors and metadata
         S-->>R: validated HostWeightLease
+        R-->>L: lease and LOCAL_PRODUCTION report
+        L->>X: plan_restore(model, lease)
+        X-->>L: validation-only restore plan
+        L->>X: commit() once
     else policy permits canonical fallback
         R-->>L: CANONICAL_FALLBACK
         L->>L: Run canonical loader
+        opt explicit post-load publication enabled
+            L->>R: publish_after_load(identity, POST_LOAD_ONLY producer)
+            R->>S: get_or_build(identity, producer)
+            S-->>R: validated HostWeightLease or typed failure
+            R->>R: close publication lease
+            R-->>L: separate publication report
+        end
     else required or nonretryable failure
         R-->>L: FAILED report
     end
-    R-->>L: lease and terminal report
-    L->>X: plan_restore(model, lease)
-    X-->>L: validation-only restore plan
-    L->>X: commit() once
-    L->>T: consume restored CPU tensors
+    L->>T: consume final CPU tensors and any lease
     T-->>L: transfer teardown complete
-    L->>L: close lease
+    L->>L: close lease when one was acquired
 ```
 
 `plan_restore()` must not mutate the model or lease. `commit() -> None` is the
@@ -143,6 +162,15 @@ sole one-shot model mutation. If planning fails, canonical fallback may reuse
 the untouched model. If commit begins and fails, the partially hydrated model
 must be discarded and canonical fallback must construct a fresh model.
 
+`publish_after_load()` is synchronous in V1 and accepts only a
+`POST_LOAD_ONLY` producer. Its report is separate from the terminal resolution,
+so a publication failure cannot rewrite a successful canonical fallback. A
+successful publication closes the store-returned lease inside the runtime and
+warms only future startups. It does not restore, rebind, or otherwise mutate the
+canonically loaded model serving the current startup.
+`allow_local_build` gates producers during pre-load resolution, while
+`allow_post_load_publish` independently gates this explicit post-load path.
+
 ## Exact representation identity
 
 The store performs exact lookup. It never silently converts one representation
@@ -252,7 +280,8 @@ removing live mappings.
 | Retryable lock, domain, or capacity failure | Canonical fallback | Fail |
 | Unsupported producer or backend | Fail | Fail |
 | Identity collision | Fail | Fail |
-| Producer or publication failure | Fail | Fail |
+| Pre-load producer or atomic publication failure | Fail | Fail |
+| Post-load publication failure after canonical fallback | Keep canonical model; report publication failure | Not reached through required resolution |
 | Restore planning failure | Canonical fallback may reuse untouched model | Fail |
 | Restore commit failure | Discard model; fallback requires a fresh instance | Fail |
 
@@ -275,7 +304,8 @@ A consumer PR must:
    a failed restore commit.
 7. Retain the lease until every transport operation that may access mapped
    memory has completed.
-8. Emit and test the terminal resolution report.
+8. Emit and test the terminal resolution report and any separate post-load
+   publication report.
 
 Consumer validation must prove:
 

docs/design/module/host_weight_runtime.md

@@ -55,7 +55,9 @@ flowchart LR
 The contracts have deliberately narrow ownership:
 
 - `HostWeightRuntime` applies resolution and fallback policy and emits one
-  terminal report. It does not own a model or call the canonical loader.
+  terminal resolution report. Explicit post-load publication emits a separate
+  report and may return a lease, but does not revise the completed resolution.
+  The runtime does not own a model or call the canonical loader.
 - `HostWeightStore` owns artifact lookup, coordinated construction,
   publication, validation, quarantine, and lifecycle.
 - `HostWeightLease` owns process-local tensor views, mapped-file resources, and
@@ -135,9 +137,23 @@ producer cancellation requires a future process-isolated producer contract. A
 waiter timing out never cancels another process's valid build.
 
 Remote providers are represented by protocols but are explicitly unsupported
-in the first implementation. V1 also invokes only `PRE_LOAD_SAFE` producers;
-enabling post-load publication is rejected during configuration until a
-separate post-load entry point and model-lifetime contract exist.
+in the first implementation. Pre-load `resolve()` invokes only
+`PRE_LOAD_SAFE` producers. A loader that completed canonical materialization
+may explicitly call `publish_after_load()` with a `POST_LOAD_ONLY` producer
+when `allow_post_load_publish` is enabled.
+
+`allow_local_build` controls producer use during pre-load resolution;
+`allow_post_load_publish` independently controls the explicit post-load entry
+point. This permits lookup-only warm starts that populate a missing artifact
+only after the canonical loader has established the current model.
+
+Post-load publication is synchronous in V1, but its outcome is independent of
+the already completed resolution. A store failure is returned in a separate
+publication report and cannot turn a valid canonical model into a failed
+resolution. On success, the runtime closes the validated store lease before
+returning the report. Post-load publication only warms future startups; it does
+not restore, rebind, or otherwise mutate the canonically loaded model serving
+the current startup.
 
 ## Restoration transaction boundary
 

tests/host_weight_runtime/test_contracts.py

@@ -142,9 +142,11 @@ def test_restoration_contract_has_a_validation_only_one_shot_commit_boundary() -
     assert "without mutating" in (WeightRestorer.__doc__ or "")
 
 
-def test_post_load_publication_policy_is_explicitly_unsupported() -> None:
-    with pytest.raises(ValueError, match="post-load publication is not implemented"):
-        ProductionPolicy(allow_post_load_publish=True)
+def test_post_load_publication_policy_is_explicitly_enabled() -> None:
+    policy = ProductionPolicy(allow_post_load_publish=True)
+
+    assert policy.allow_local_build
+    assert policy.allow_post_load_publish
 
 
 def test_manifest_and_publication_marker_round_trip_exactly() -> None:

tests/host_weight_runtime/test_runtime_resolution.py

@@ -1,10 +1,11 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-"""Resolution-policy tests for the loader-adjacent Host Weight Runtime."""
+"""Runtime resolution and publication-policy tests for Host Weight Runtime."""
 
 from __future__ import annotations
 
 from pathlib import Path
+from typing import cast
 
 import pytest
 import torch
@@ -15,12 +16,16 @@ from vllm_omni.host_weight_runtime import (
     ComponentIdentity,
     FailureCode,
     HostWeightFailure,
+    HostWeightLease,
     HostWeightRuntime,
     HostWeightRuntimeConfig,
     IntegrityPolicy,
     LookupPhase,
+    PostLoadPublicationOutcome,
+    PostLoadPublicationReport,
     ProducerIdentity,
     ProductionMetadata,
+    ProductionPolicy,
     ProductionSourceMode,
     RemoteImportPolicy,
     RemoteOnMiss,
@@ -39,7 +44,7 @@ from vllm_omni.host_weight_runtime import (
     WeightRepresentation,
     WeightSourceIdentity,
 )
-from vllm_omni.host_weight_runtime.filesystem import detect_storage_class
+from vllm_omni.host_weight_runtime.filesystem import FilesystemHostWeightStore, detect_storage_class
 
 pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
 
@@ -224,6 +229,246 @@ def test_preload_resolution_does_not_run_postload_only_producer(tmp_path: Path)
     assert resolution.report.attempts[0].action.value == "canonical_fallback"
 
 
+def test_postload_publication_closes_lease_and_only_warms_future_resolution(
+    tmp_path: Path,
+) -> None:
+    publications: list[PostLoadPublicationReport] = []
+    identity = _identity()
+    producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.PREFERRED,
+            domain=_domain(tmp_path / "store"),
+            production=ProductionPolicy(
+                allow_local_build=False,
+                allow_post_load_publish=True,
+            ),
+        ),
+        publication_observer=publications.append,
+    )
+
+    preload = runtime.resolve(identity, producer=producer)
+    assert preload.report.outcome is ResolutionOutcome.CANONICAL_FALLBACK
+    assert producer.calls == 0
+
+    publication = runtime.publish_after_load(identity, producer=producer)
+    assert publication.outcome is PostLoadPublicationOutcome.PUBLISHED
+    assert publication.identity_digest == identity.key
+
+    already_present = runtime.publish_after_load(identity, producer=producer)
+    assert already_present.outcome is PostLoadPublicationOutcome.ALREADY_PRESENT
+
+    warm = runtime.resolve(identity)
+    assert warm.report.outcome is ResolutionOutcome.LOCAL_HIT
+    assert warm.lease is not None
+    assert torch.equal(
+        warm.lease.tensors["weight"],
+        torch.arange(4, dtype=torch.float32).to(torch.bfloat16).reshape(2, 2),
+    )
+    warm.lease.close()
+    assert producer.calls == 1
+    assert publications == [publication, already_present]
+    assert isinstance(runtime.store, FilesystemHostWeightStore)
+    assert runtime.store.cleanup(identity) is None
+
+
+def test_postload_publication_policy_skip_does_not_run_producer(tmp_path: Path) -> None:
+    identity = _identity()
+    producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(mode=RuntimeMode.PREFERRED, domain=_domain(tmp_path / "store"))
+    )
+
+    publication = runtime.publish_after_load(identity, producer=producer)
+
+    assert publication.outcome is PostLoadPublicationOutcome.POLICY_DISABLED
+    assert producer.calls == 0
+
+
+def test_postload_publication_reports_lease_cleanup_failure(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    identity = _identity()
+    producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.PREFERRED,
+            domain=_domain(tmp_path / "store"),
+            production=ProductionPolicy(allow_post_load_publish=True),
+        )
+    )
+    assert runtime.store is not None
+
+    class FailingLease:
+        def close(self) -> None:
+            raise RuntimeError("injected lease cleanup failure")
+
+    def return_failing_lease(*_args: object, **_kwargs: object) -> StoreResult:
+        return StoreResult(StoreStatus.BUILT, lease=cast(HostWeightLease, FailingLease()))
+
+    monkeypatch.setattr(runtime.store, "get_or_build", return_failing_lease)
+
+    publication = runtime.publish_after_load(identity, producer=producer)
+
+    assert publication.outcome is PostLoadPublicationOutcome.FAILED
+    assert publication.failure is not None
+    assert publication.failure.stage is ResolutionStage.LIFECYCLE
+    assert publication.failure.code is FailureCode.PUBLICATION_FAILED
+    assert publication.failure.details.to_value() == {
+        "exception_type": "RuntimeError",
+        "store_outcome": "built",
+    }
+
+
+def test_disabled_postload_publication_does_not_probe_configured_root(tmp_path: Path) -> None:
+    root = tmp_path / "must-not-exist"
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.DISABLED,
+            domain=_domain(root),
+            production=ProductionPolicy(allow_post_load_publish=True),
+        )
+    )
+
+    publication = runtime.publish_after_load()
+
+    assert publication.outcome is PostLoadPublicationOutcome.RUNTIME_DISABLED
+    assert not root.exists()
+
+
+def test_postload_publication_rejects_preload_producer(tmp_path: Path) -> None:
+    identity = _identity()
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.PREFERRED,
+            domain=_domain(tmp_path / "store"),
+            production=ProductionPolicy(allow_post_load_publish=True),
+        )
+    )
+
+    with pytest.raises(ValueError, match="POST_LOAD_ONLY"):
+        runtime.publish_after_load(identity, producer=CountingProducer(identity))
+
+
+def test_postload_publication_failure_is_reported_without_raising(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    identity = _identity()
+    producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.PREFERRED,
+            domain=_domain(tmp_path / "store"),
+            production=ProductionPolicy(allow_post_load_publish=True),
+        )
+    )
+    assert runtime.store is not None
+    failure = HostWeightFailure(
+        stage=ResolutionStage.CAPACITY,
+        code=FailureCode.ENOSPC,
+        retryable=True,
+        message="injected post-load publication failure",
+        details=CanonicalJson.empty(),
+    )
+
+    def fail_publication(*_args: object, **_kwargs: object) -> StoreResult:
+        return StoreResult(StoreStatus.FAILED, failure=failure)
+
+    monkeypatch.setattr(runtime.store, "get_or_build", fail_publication)
+    publication = runtime.publish_after_load(identity, producer=producer)
+
+    assert publication.outcome is PostLoadPublicationOutcome.FAILED
+    assert publication.failure is failure
+    assert producer.calls == 0
+
+
+def test_postload_publication_reports_store_initialization_failure(tmp_path: Path) -> None:
+    root = tmp_path / "not-a-directory"
+    root.write_text("occupied", encoding="utf-8")
+    identity = _identity()
+    producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.PREFERRED,
+            domain=_domain(root),
+            production=ProductionPolicy(allow_post_load_publish=True),
+        )
+    )
+
+    publication = runtime.publish_after_load(identity, producer=producer)
+
+    assert publication.outcome is PostLoadPublicationOutcome.FAILED
+    assert publication.failure is not None
+    assert publication.failure.code is FailureCode.DOMAIN_UNAVAILABLE
+    assert publication.failure.retryable
+    assert producer.calls == 0
+
+
+def test_postload_publication_maps_joined_and_closes_lease(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    identity = _identity()
+    producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.PREFERRED,
+            domain=_domain(tmp_path / "store"),
+            production=ProductionPolicy(allow_post_load_publish=True),
+        )
+    )
+    assert runtime.store is not None
+
+    class ClosingLease:
+        closed = False
+
+        def close(self) -> None:
+            self.closed = True
+
+    lease = ClosingLease()
+
+    def return_joined_lease(*_args: object, **_kwargs: object) -> StoreResult:
+        return StoreResult(StoreStatus.JOINED, lease=cast(HostWeightLease, lease))
+
+    monkeypatch.setattr(runtime.store, "get_or_build", return_joined_lease)
+
+    publication = runtime.publish_after_load(identity, producer=producer)
+
+    assert publication.outcome is PostLoadPublicationOutcome.JOINED
+    assert lease.closed
+
+
+def test_postload_publication_synthesizes_failure_for_unexpected_store_status(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    identity = _identity()
+    producer = CountingProducer(identity, lookup_phase=LookupPhase.POST_LOAD_ONLY)
+    runtime = HostWeightRuntime.from_config(
+        HostWeightRuntimeConfig(
+            mode=RuntimeMode.PREFERRED,
+            domain=_domain(tmp_path / "store"),
+            production=ProductionPolicy(allow_post_load_publish=True),
+        )
+    )
+    assert runtime.store is not None
+
+    def return_unexpected_miss(*_args: object, **_kwargs: object) -> StoreResult:
+        return StoreResult(StoreStatus.MISS)
+
+    monkeypatch.setattr(runtime.store, "get_or_build", return_unexpected_miss)
+
+    publication = runtime.publish_after_load(identity, producer=producer)
+
+    assert publication.outcome is PostLoadPublicationOutcome.FAILED
+    assert publication.failure is not None
+    assert publication.failure.stage is ResolutionStage.PRODUCTION
+    assert publication.failure.code is FailureCode.PRODUCER_FAILED
+    assert not publication.failure.retryable
+
+
 @pytest.mark.parametrize(
     ("mode", "expected"),
     [

vllm_omni/host_weight_runtime/__init__.py

@@ -47,6 +47,8 @@ from .manifest import (
 from .outcomes import (
     AttemptResult,
     HostWeightResolution,
+    PostLoadPublicationOutcome,
+    PostLoadPublicationReport,
     ResolutionAttempt,
     ResolutionOutcome,
     ResolutionReport,
@@ -102,6 +104,8 @@ __all__ = [
     "ManifestValidationError",
     "MappedHostRegion",
     "ProducerIdentity",
+    "PostLoadPublicationOutcome",
+    "PostLoadPublicationReport",
     "ProductionMetadata",
     "ProductionPolicy",
     "ProductionSourceMode",

vllm_omni/host_weight_runtime/config.py

@@ -117,8 +117,6 @@ class ProductionPolicy:
     def __post_init__(self) -> None:
         if not isinstance(self.allow_local_build, bool) or not isinstance(self.allow_post_load_publish, bool):
             raise ValueError("production switches must be booleans")
-        if self.allow_post_load_publish:
-            raise ValueError("post-load publication is not implemented")
 
 
 @dataclass(frozen=True)

vllm_omni/host_weight_runtime/outcomes.py

@@ -1,6 +1,6 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
-"""Typed store and end-to-end resolution outcomes."""
+"""Typed store, resolution, and post-load publication outcomes."""
 
 from __future__ import annotations
 
@@ -60,6 +60,17 @@ class ResolutionOutcome(str, Enum):
     FAILED = "failed"
 
 
+class PostLoadPublicationOutcome(str, Enum):
+    """Terminal outcome for one explicit post-load publication attempt."""
+
+    RUNTIME_DISABLED = "runtime_disabled"
+    POLICY_DISABLED = "policy_disabled"
+    ALREADY_PRESENT = "already_present"
+    PUBLISHED = "published"
+    JOINED = "joined"
+    FAILED = "failed"
+
+
 @dataclass(frozen=True)
 class ResolutionAttempt:
     stage: ResolutionStage
@@ -108,9 +119,38 @@ class HostWeightResolution:
             raise ValueError("resolution outcome has inconsistent lease ownership")
 
 
+@dataclass(frozen=True)
+class PostLoadPublicationReport:
+    operation_id: str
+    outcome: PostLoadPublicationOutcome
+    identity_digest: str | None
+    elapsed_seconds: float
+    failure: HostWeightFailure | None = None
+
+    def __post_init__(self) -> None:
+        if not self.operation_id:
+            raise ValueError("post-load publication report requires an operation_id")
+        if not isinstance(self.outcome, PostLoadPublicationOutcome):
+            raise ValueError("post-load publication outcome must use PostLoadPublicationOutcome")
+        if self.identity_digest is not None and not self.identity_digest:
+            raise ValueError("post-load publication identity digest must not be empty")
+        requires_identity = self.outcome not in {
+            PostLoadPublicationOutcome.RUNTIME_DISABLED,
+            PostLoadPublicationOutcome.POLICY_DISABLED,
+        }
+        if requires_identity != (self.identity_digest is not None):
+            raise ValueError("post-load publication identity ownership is inconsistent")
+        if not math.isfinite(self.elapsed_seconds) or self.elapsed_seconds < 0:
+            raise ValueError("post-load publication elapsed_seconds must be finite and non-negative")
+        if (self.outcome is PostLoadPublicationOutcome.FAILED) != (self.failure is not None):
+            raise ValueError("post-load publication failure ownership is inconsistent")
+
+
 __all__ = [
     "AttemptResult",
     "HostWeightResolution",
+    "PostLoadPublicationOutcome",
+    "PostLoadPublicationReport",
     "ResolutionAttempt",
     "ResolutionOutcome",
     "ResolutionReport",

vllm_omni/host_weight_runtime/runtime.py

@@ -16,6 +16,8 @@ from .lease import HostWeightLease
 from .outcomes import (
     AttemptResult,
     HostWeightResolution,
+    PostLoadPublicationOutcome,
+    PostLoadPublicationReport,
     ResolutionAttempt,
     ResolutionOutcome,
     ResolutionReport,
@@ -28,7 +30,7 @@ logger = logging.getLogger(__name__)
 
 
 class HostWeightRuntime:
-    """Resolve one exact representation without owning model restoration."""
+    """Resolve or publish one exact representation without owning restoration."""
 
     def __init__(
         self,
@@ -36,10 +38,13 @@ class HostWeightRuntime:
         store: HostWeightStore | None = None,
         *,
         observer: Callable[[ResolutionReport], None] | None = None,
+        publication_observer: Callable[[PostLoadPublicationReport], None] | None = None,
         _initialization_failure: HostWeightFailure | None = None,
     ) -> None:
         if observer is not None and not callable(observer):
             raise ValueError("Host Weight Runtime observer must be callable")
+        if publication_observer is not None and not callable(publication_observer):
+            raise ValueError("Host Weight Runtime publication observer must be callable")
         if config.mode is RuntimeMode.DISABLED and (store is not None or _initialization_failure is not None):
             raise ValueError("disabled Host Weight Runtime must not initialize a store")
         if config.mode is not RuntimeMode.DISABLED and (store is None) == (_initialization_failure is None):
@@ -48,6 +53,7 @@ class HostWeightRuntime:
         self.store = store
         self._initialization_failure = _initialization_failure
         self._observer = observer
+        self._publication_observer = publication_observer
 
     @classmethod
     def from_config(
@@ -55,18 +61,24 @@ class HostWeightRuntime:
         config: HostWeightRuntimeConfig,
         *,
         observer: Callable[[ResolutionReport], None] | None = None,
+        publication_observer: Callable[[PostLoadPublicationReport], None] | None = None,
     ) -> HostWeightRuntime:
         """Construct lazily so disabled mode performs no filesystem probe."""
         if config.mode is RuntimeMode.DISABLED:
-            return cls(config, observer=observer)
+            return cls(config, observer=observer, publication_observer=publication_observer)
         assert config.domain is not None
         from .filesystem import FilesystemHostWeightStore  # noqa: PLC0415
 
         try:
             store = FilesystemHostWeightStore(config.domain, config.capacity, config.integrity)
         except HostWeightError as exc:
-            return cls(config, observer=observer, _initialization_failure=exc.failure)
-        return cls(config, store, observer=observer)
+            return cls(
+                config,
+                observer=observer,
+                publication_observer=publication_observer,
+                _initialization_failure=exc.failure,
+            )
+        return cls(config, store, observer=observer, publication_observer=publication_observer)
 
     @property
     def enabled(self) -> bool:
@@ -223,6 +235,123 @@ class HostWeightRuntime:
 
         return self._finish_terminal(tuple(attempts), started)
 
+    def publish_after_load(
+        self,
+        identity: WeightArtifactIdentity | None = None,
+        *,
+        producer: WeightProducer | None = None,
+    ) -> PostLoadPublicationReport:
+        """Synchronously warm a future startup without revising resolution.
+
+        A successful store operation returns a validated lease. The runtime
+        closes that lease before returning because POST_LOAD_ONLY production
+        must not mutate or rebind the model serving the current startup.
+        """
+        started = time.monotonic()
+        operation_id = uuid.uuid4().hex
+        if not self.enabled:
+            return self._finish_publication(
+                PostLoadPublicationReport(
+                    operation_id=operation_id,
+                    outcome=PostLoadPublicationOutcome.RUNTIME_DISABLED,
+                    identity_digest=None,
+                    elapsed_seconds=time.monotonic() - started,
+                )
+            )
+        if not self.config.production.allow_post_load_publish:
+            return self._finish_publication(
+                PostLoadPublicationReport(
+                    operation_id=operation_id,
+                    outcome=PostLoadPublicationOutcome.POLICY_DISABLED,
+                    identity_digest=None,
+                    elapsed_seconds=time.monotonic() - started,
+                )
+            )
+        if identity is None:
+            raise ValueError("enabled post-load publication requires an exact artifact identity")
+        if producer is None:
+            raise ValueError("enabled post-load publication requires one producer")
+        if producer.spec.lookup_phase is not LookupPhase.POST_LOAD_ONLY:
+            raise ValueError("publish_after_load requires a POST_LOAD_ONLY producer")
+
+        identity_digest = identity.key
+        if self._initialization_failure is not None:
+            return self._finish_publication(
+                PostLoadPublicationReport(
+                    operation_id=operation_id,
+                    outcome=PostLoadPublicationOutcome.FAILED,
+                    identity_digest=identity_digest,
+                    elapsed_seconds=time.monotonic() - started,
+                    failure=self._initialization_failure,
+                )
+            )
+
+        assert self.store is not None
+        result = self.store.get_or_build(
+            BuildRequest(identity),
+            producer,
+            validation=self.config.integrity.local_lookup,
+            deadline=started + self.config.wait.coordination_timeout_seconds,
+        )
+        successful_outcomes = {
+            StoreStatus.HIT: PostLoadPublicationOutcome.ALREADY_PRESENT,
+            StoreStatus.BUILT: PostLoadPublicationOutcome.PUBLISHED,
+            StoreStatus.JOINED: PostLoadPublicationOutcome.JOINED,
+        }
+        outcome = successful_outcomes.get(result.status)
+        if outcome is not None:
+            assert result.lease is not None
+            try:
+                result.lease.close()
+            except Exception as exc:
+                return self._finish_publication(
+                    PostLoadPublicationReport(
+                        operation_id=operation_id,
+                        outcome=PostLoadPublicationOutcome.FAILED,
+                        identity_digest=identity_digest,
+                        elapsed_seconds=time.monotonic() - started,
+                        failure=HostWeightFailure(
+                            stage=ResolutionStage.LIFECYCLE,
+                            code=FailureCode.PUBLICATION_FAILED,
+                            retryable=False,
+                            message="post-load publication lease cleanup failed",
+                            details=CanonicalJson.from_value(
+                                {
+                                    "exception_type": type(exc).__name__,
+                                    "store_outcome": result.status.value,
+                                }
+                            ),
+                        ),
+                    )
+                )
+            return self._finish_publication(
+                PostLoadPublicationReport(
+                    operation_id=operation_id,
+                    outcome=outcome,
+                    identity_digest=identity_digest,
+                    elapsed_seconds=time.monotonic() - started,
+                )
+            )
+
+        failure = result.failure
+        if failure is None:
+            failure = HostWeightFailure(
+                stage=ResolutionStage.PRODUCTION,
+                code=FailureCode.PRODUCER_FAILED,
+                retryable=False,
+                message=f"post-load publication returned unexpected store status {result.status.value}",
+                details=CanonicalJson.empty(),
+            )
+        return self._finish_publication(
+            PostLoadPublicationReport(
+                operation_id=operation_id,
+                outcome=PostLoadPublicationOutcome.FAILED,
+                identity_digest=identity_digest,
+                elapsed_seconds=time.monotonic() - started,
+                failure=failure,
+            )
+        )
+
     def _mode_terminal_action(self) -> ResolutionAction:
         if self.config.mode is RuntimeMode.PREFERRED:
             return ResolutionAction.CANONICAL_FALLBACK
@@ -328,5 +457,19 @@ class HostWeightRuntime:
                 )
         return HostWeightResolution(report=report, lease=lease)
 
+    def _finish_publication(
+        self,
+        report: PostLoadPublicationReport,
+    ) -> PostLoadPublicationReport:
+        if self._publication_observer is not None:
+            try:
+                self._publication_observer(report)
+            except Exception:
+                logger.exception(
+                    "Host Weight Runtime publication observer failed",
+                    extra={"operation_id": report.operation_id},
+                )
+        return report
+
 
 __all__ = ["HostWeightRuntime"]