Load read-only manifested snapshots with separate writable caches

.github/workflows/run-tests.yaml

@@ -112,6 +112,7 @@ jobs:
             tests/test_model_catalog.py \
             tests/test_model_manager.py \
             tests/test_model_manifest.py \
+            tests/test_local_artifact_snapshot.py \
             tests/test_multi_machine_qualification.py \
             tests/test_native_credentials.py \
             tests/test_node_config.py \

docs/MODEL_QUALIFICATION_V1.md

@@ -61,6 +61,45 @@ with opaque labels. The report also records the normalized operating system, req
 device profile, observed worker device, dtype, attention implementation, and the source
 commit when it can resolve one from the checkout.
 
+## Loading an existing read-only snapshot
+
+The fixed-manifest worker and text-peer CLIs accept `--artifact_root` for a
+complete local snapshot, paired with an explicit, separate writable `--cache_dir`:
+
+```text
+drift server <repository> --model_manifest <manifest.json> \
+  --artifact_root /models/verified-snapshot --cache_dir /run/communityai/cache \
+  --block_indices 0:14 --identity_path <owned-worker-identity> <other-worker-options>
+drift text-peer <manifest.json> \
+  --artifact_root /models/verified-snapshot --cache_dir /run/communityai/cache \
+  --identity_path <owned-text-identity> --initial_peers <owned-bootstrap-address>
+```
+
+Each declared artifact lives at its manifest-relative path below `artifact_root`.
+This interface requires a materialized tree containing exactly the declared
+regular files and their parent directories. Symlinks, Windows reparse points,
+extra files (including loader sidecars, README/license files) and extra directories
+are rejected. A conventional Hub snapshot containing blob symlinks therefore
+needs a separate materialized, manifest-only snapshot for this interface.
+Every declared file, including unassigned shards and tokenizer files, is checked
+by size and SHA-256 before construction of a model loader; the text-peer CLI also
+checks before identity/DHT startup. Missing or corrupt files fail without Hub or
+custom HTTP fallback. The root reaches worker config, module and block loaders,
+and the text-peer client loader. Runtime locks such as `blocks.lock` belong to
+`cache_dir`; neither directory may contain the other. The caller supplies and
+maintains the read-only snapshot mount. Managed workers retain all five manifest,
+span, artifact-byte, artifact-set and writable-cache placement claims.
+These examples apply to direct `drift server` and `drift text-peer` launches.
+The `drift node` supervisor does not propagate this opt-in flag to its children.
+
+The Python `make_manifest_loader(..., artifact_root=..., cache_dir=...)` interface
+uses the same boundary. `ManifestArtifactVerifier(..., cache_only=True)` separately
+supports verified existing Hub/custom cache entries without starting a transfer
+on a miss. Omitting these opt-in settings preserves the existing online behavior.
+Content validation does not create signed catalog approval or authorize model or
+provider execution. Direct fixed-manifest experiments and automatic signed-catalog
+selection remain separate interfaces.
+
 ## Cross-platform matrix gate
 
 Each real host must produce a complete artifact, parity, and local-failover report with

src/drift/artifact_snapshot.py

@@ -0,0 +1,72 @@
+"""Local artifact snapshot validation, independent of catalog or owner admission."""
+
+import stat
+from pathlib import Path, PurePosixPath
+from typing import Optional, Union
+
+from drift.model_manifest import ManifestError, ModelManifest
+
+
+def validate_artifact_snapshot(
+    manifest: Optional[ModelManifest],
+    artifact_root: Optional[Union[str, Path]],
+    *,
+    cache_dir: Optional[Union[str, Path]],
+) -> Optional[Path]:
+    """Hash a materialized, manifest-only tree without writing to the snapshot.
+
+    Runtime locks belong to an explicit, separate cache directory. This checks
+    content only; it does not approve a catalog entry or authorize a role start.
+    """
+    if artifact_root is None:
+        return None
+    if not isinstance(manifest, ModelManifest):
+        raise ManifestError("artifact_root requires an exact model manifest")
+    if cache_dir is None or not str(cache_dir):
+        raise ManifestError("artifact_root requires a separate writable cache_dir")
+    try:
+        root = Path(artifact_root).expanduser().resolve(strict=True)
+        cache = Path(cache_dir).expanduser().resolve()
+    except (OSError, TypeError, ValueError) as exc:
+        raise ManifestError("Could not resolve artifact snapshot or cache directory") from exc
+    if not root.is_dir():
+        raise ManifestError("artifact_root must be an existing snapshot directory")
+    if cache.exists() and not cache.is_dir():
+        raise ManifestError("cache_dir must be a writable cache directory")
+    if root == cache or root.is_relative_to(cache) or cache.is_relative_to(root):
+        raise ManifestError("artifact_root and writable cache_dir must be disjoint directories")
+    _validate_snapshot_tree(manifest, root)
+    manifest.verify_artifacts(root)
+    return root
+
+
+def _validate_snapshot_tree(manifest: ModelManifest, root: Path) -> None:
+    files = {artifact.path for artifact in manifest.artifacts}
+    directories = {
+        parent.as_posix()
+        for filename in files
+        for parent in PurePosixPath(filename).parents
+        if parent != PurePosixPath(".")
+    }
+    pending = [root]
+    try:
+        while pending:
+            for path in pending.pop().iterdir():
+                relative = path.relative_to(root).as_posix()
+                metadata = path.lstat()
+                if stat.S_ISLNK(metadata.st_mode) or getattr(metadata, "st_file_attributes", 0) & getattr(
+                    stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0
+                ):
+                    raise ManifestError(f"Artifact snapshot cannot contain links or reparse points: {relative}")
+                if not path.resolve(strict=True).is_relative_to(root):
+                    raise ManifestError(f"Artifact escapes snapshot root: {relative}")
+                if stat.S_ISDIR(metadata.st_mode):
+                    if relative not in directories:
+                        raise ManifestError(f"Undeclared artifact snapshot directory: {relative}")
+                    pending.append(path)
+                elif not stat.S_ISREG(metadata.st_mode):
+                    raise ManifestError(f"Artifact snapshot entry must be a regular file: {relative}")
+                elif relative not in files:
+                    raise ManifestError(f"Undeclared artifact snapshot file: {relative}")
+    except OSError as exc:
+        raise ManifestError("Could not inspect artifact snapshot tree") from exc

src/drift/cli/run_server.py

@@ -250,6 +250,8 @@ def build_parser(*, bound_worker: bool = False) -> configargparse.ArgParser:
 
     parser.add_argument('--cache_dir', type=str, default=None,
                         help='Path to a directory in which a downloaded pretrained model configuration should be cached if the standard cache should not be used.')
+    parser.add_argument('--artifact_root', type=str, default=None,
+                        help='Read-only snapshot of every manifested artifact; requires --model_manifest and a separate writable --cache_dir. Missing or corrupt files fail without download.')
     parser.add_argument("--max_disk_space", type=str, default=None,
                         help="Maximal disk space used for caches. Example: 50GB, 100GiB (GB != GiB here). "
                              "Default: unlimited. "

src/drift/cli/run_text_peer.py

@@ -7,6 +7,7 @@
 
 from hivemind import DHT
 
+from drift.artifact_snapshot import validate_artifact_snapshot
 from drift.model_manifest import ModelManifest
 from drift.protocol_identity import NodeIdentity
 from drift.server.text_peer import TextPeerService
@@ -18,12 +19,16 @@ def main():
     parser.add_argument("--initial_peers", nargs="+", required=True)
     parser.add_argument("--identity_path", type=Path, required=True)
     parser.add_argument("--cache_dir", type=Path, required=True)
+    parser.add_argument(
+        "--artifact_root", type=Path, help="Complete read-only manifested snapshot; cache_dir owns locks"
+    )
     parser.add_argument("--host_maddrs", nargs="+", default=["/ip4/0.0.0.0/tcp/31337"])
     parser.add_argument("--announce_maddrs", nargs="+")
     parser.add_argument("--max_context_tokens", type=int, default=2048)
     parser.add_argument("--max_output_tokens", type=int, default=512)
     args = parser.parse_args()
     manifest = ModelManifest.load(args.manifest)
+    artifact_root = validate_artifact_snapshot(manifest, args.artifact_root, cache_dir=args.cache_dir)
     identity = NodeIdentity.ensure(args.identity_path)
     dht = DHT(
         initial_peers=args.initial_peers,
@@ -39,6 +44,7 @@ def main():
         manifest,
         initial_peers=args.initial_peers,
         cache_dir=str(args.cache_dir),
+        artifact_root=artifact_root,
         max_context_tokens=args.max_context_tokens,
         max_output_tokens=args.max_output_tokens,
     ).start()

src/drift/model_manifest.py

@@ -661,7 +661,8 @@ class ManifestArtifactVerifier:
     A verifier deliberately materializes only requested files. This preserves Petals' partial-checkpoint
     behavior: a worker verifies the shards containing its assigned blocks, while an API client verifies the
     tokenizer and the shards containing its local embeddings/head. Successful hashes are cached only while
-    the resolved path, size, and modification timestamp remain unchanged.
+    the resolved path, size, and modification timestamp remain unchanged. Explicit ``cache_only`` mode
+    accepts verified local files but never starts a transfer on a cache miss.
     """
 
     manifest: ModelManifest
@@ -672,6 +673,7 @@ class ManifestArtifactVerifier:
     max_disk_space: Optional[int] = None
     artifact_root: Optional[Union[str, os.PathLike]] = None
     allowed_paths: Optional[Iterable[str]] = None
+    cache_only: bool = False
     _verified: Dict[Tuple[str, int, int, str], bool] = field(default_factory=dict, init=False, repr=False)
     _snapshot_root: Optional[Path] = field(default=None, init=False, repr=False)
     _weight_map: Optional[Mapping[str, str]] = field(default=None, init=False, repr=False)
@@ -690,6 +692,8 @@ def __post_init__(self) -> None:
             raise ManifestError(
                 f"Artifact verifier revision is {self.revision!r}, expected {self.manifest.source.revision!r}"
             )
+        if type(self.cache_only) is not bool:
+            raise ManifestError("Artifact verifier cache_only must be a boolean")
         if self.allowed_paths is not None:
             self.allowed_paths = self._normalize_allowed_paths(self.allowed_paths)
         if self.artifact_root is not None:
@@ -842,6 +846,7 @@ def _ensure_path(self, path: str, *, allowed_roles: Optional[Iterable[str]] = No
             )
 
         candidate = None
+        already_verified = False
         if self._snapshot_root is not None:
             candidate = _artifact_path_below_root(self._snapshot_root, artifact.path)
         if candidate is None or (not candidate.exists() and self.artifact_root is None):
@@ -862,16 +867,24 @@ def _ensure_path(self, path: str, *, allowed_roles: Optional[Iterable[str]] = No
                     resolved = None
 
                 if resolved is None:
-                    from drift.utils.disk_cache import allow_cache_writes, free_disk_space_for
-
-                    Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
-                    with allow_cache_writes(self.cache_dir):
-                        free_disk_space_for(
-                            max(0, artifact.size - self.partial_size(artifact.path)),
-                            cache_dir=self.cache_dir,
-                            max_disk_space=self.max_disk_space,
-                        )
-                        resolved = self._resumable_hub_download(artifact, destination=candidate)
+                    if self.cache_only:
+                        _, cached_final, _ = self._resumable_paths(artifact)
+                        if not cached_final.exists():
+                            raise ManifestError(f"Artifact {artifact.path} is not available in the local cache")
+                        _verify_artifact_file(artifact, cached_final)
+                        already_verified = True
+                        resolved = str(cached_final)
+                    else:
+                        from drift.utils.disk_cache import allow_cache_writes, free_disk_space_for
+
+                        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
+                        with allow_cache_writes(self.cache_dir):
+                            free_disk_space_for(
+                                max(0, artifact.size - self.partial_size(artifact.path)),
+                                cache_dir=self.cache_dir,
+                                max_disk_space=self.max_disk_space,
+                            )
+                            resolved = self._resumable_hub_download(artifact, destination=candidate)
             except ManifestTransferInterrupted:
                 raise
             except Exception as exc:
@@ -893,7 +906,8 @@ def _ensure_path(self, path: str, *, allowed_roles: Optional[Iterable[str]] = No
                     self._promote_cached_artifact(artifact, resolved_candidate, candidate)
 
         self._report(artifact, "verifying", received=artifact.size)
-        self._verify(artifact, candidate)
+        if not already_verified:
+            self._verify(artifact, candidate)
         return candidate
 
     def _promote_cached_artifact(self, artifact: ManifestArtifact, source: Path, destination: Path) -> None:
@@ -954,6 +968,9 @@ def _resumable_hub_download(self, artifact: ManifestArtifact, *, destination: Op
         A partial is never exposed to Transformers.  It is atomically promoted only after
         its exact declared size and SHA-256 pass.
         """
+        if self.cache_only:
+            raise ManifestError("Artifact transfer is disabled in cache-only mode")
+
         import requests
         from huggingface_hub import hf_hub_url
         from huggingface_hub.utils import build_hf_headers
@@ -1096,11 +1113,12 @@ def _verify(self, artifact: ManifestArtifact, candidate: Path) -> None:
         except OSError as exc:
             raise ManifestError(f"Could not read artifact {artifact.path}: {exc}") from exc
         cache_key = (str(candidate), stat_result.st_size, stat_result.st_mtime_ns, artifact.sha256)
-        if cache_key in self._verified:
+        if not self.cache_only and cache_key in self._verified:
             return
         _verify_artifact_file(artifact, candidate)
-        self._verified = {key: value for key, value in self._verified.items() if key[0] != str(candidate)}
-        self._verified[cache_key] = True
+        if not self.cache_only:
+            self._verified = {key: value for key, value in self._verified.items() if key[0] != str(candidate)}
+            self._verified[cache_key] = True
 
 
 def _select_snapshot_artifacts(files: Iterable[str], root: Path | str) -> Dict[str, str]:

src/drift/node/loading.py

@@ -8,6 +8,7 @@
 
 from hivemind.utils.logging import get_logger
 
+from drift.artifact_snapshot import validate_artifact_snapshot
 from drift.model_manifest import ManifestArtifactVerifier, ManifestError, ModelManifest
 from drift.node.model_manager import ModelRuntime
 from drift.node.route_health import sequence_manager_route_health
@@ -63,13 +64,15 @@ def make_manifest_loader(
     initial_peers: Sequence[str],
     token: Optional[str] = None,
     cache_dir: Optional[str] = None,
+    artifact_root: Optional[str] = None,
     revocation_files: Sequence[str] = (),
     request_timeout: float = 30,
     max_retries: int = 3,
 ) -> Callable[[], ModelRuntime]:
     """Return a lazy loader pinned to one exact manifest and swarm namespace."""
 
     def load() -> ModelRuntime:
+        root = validate_artifact_snapshot(manifest, artifact_root, cache_dir=cache_dir)
         import torch
         from transformers import AutoTokenizer
 
@@ -81,6 +84,8 @@ def load() -> ModelRuntime:
             revision=manifest.source.revision,
             token=token,
             cache_dir=cache_dir,
+            artifact_root=root,
+            cache_only=root is not None,
         )
         verifier.ensure_startup_metadata(include_tokenizer=True)
         logger.info(f"Loading tokenizer and client-side weights for {manifest.digest_id}")

src/drift/server/server.py

@@ -22,6 +22,7 @@
 from transformers import PretrainedConfig
 
 import drift
+from drift.artifact_snapshot import validate_artifact_snapshot
 from drift.constants import DTYPE_MAP
 from drift.data_structures import CHAIN_DELIMITER, UID_DELIMITER, ModelInfo, ServerInfo, ServerState, parse_uid
 from drift.model_manifest import ManifestArtifactVerifier, ManifestError, ModelManifest
@@ -181,6 +182,7 @@ def _scoped_manifest_artifact_verifier(
         cache_dir=cache_dir,
         max_disk_space=max_disk_space,
         artifact_root=artifact_root,
+        cache_only=artifact_root is not None,
         allowed_paths=startup_paths,
     )
     artifact_plan = verifier.bind_block_artifact_plan(
@@ -257,6 +259,7 @@ def __init__(
         admission_policy: Optional[AdmissionPolicy] = None,
         revocation_files: Sequence[str] = (),
         cache_dir: Optional[str] = None,
+        artifact_root: Optional[str] = None,
         max_disk_space: Optional[int] = None,
         max_device_memory: Optional[int] = None,
         max_processing_percent: float = 100,
@@ -292,6 +295,7 @@ def __init__(
         self._managed_loading_session = managed_loading_session
         self._managed_lifecycle_started = False
         self._managed_ready = False
+        self.artifact_root = validate_artifact_snapshot(model_manifest, artifact_root, cache_dir=cache_dir)
         converted_model_name_or_path = get_compatible_model_repo(converted_model_name_or_path)
         self.converted_model_name_or_path = converted_model_name_or_path
 
@@ -315,6 +319,8 @@ def __init__(
                 token=token,
                 cache_dir=cache_dir,
                 max_disk_space=max_disk_space,
+                artifact_root=self.artifact_root,
+                cache_only=self.artifact_root is not None,
             )
             config_source = artifact_verifier.ensure_startup_metadata()
         else:
@@ -732,6 +738,7 @@ def _create_module_container(self, block_indices: List[int]) -> ModuleContainer:
             inference_max_length=self.inference_max_length,
             torch_dtype=self.torch_dtype,
             cache_dir=self.cache_dir,
+            artifact_root=self.artifact_root,
             max_disk_space=self.max_disk_space,
             device=self.device,
             compression=self.compression,
@@ -919,6 +926,7 @@ def create(
         torch_dtype: torch.dtype,
         cache_dir: str,
         max_disk_space: int,
+        artifact_root: Optional[str] = None,
         device: Union[str, torch.device],
         compression: CompressionType,
         update_period: float,
@@ -954,6 +962,7 @@ def create(
             expected_artifact_bytes=expected_artifact_bytes,
             expected_artifact_set_digest=expected_artifact_set_digest,
             expected_cache_root=expected_cache_root,
+            artifact_root=validate_artifact_snapshot(model_manifest, artifact_root, cache_dir=cache_dir),
         )
         memory_cache = MemoryCache(attn_cache_bytes, max_alloc_timeout, paged=paged_cache, page_size=page_size)