feat(storage): add portable session artifact gateway

docs/09_OBSERVATION_SESSIONS.md

@@ -230,6 +230,13 @@ export MISSIONCORE_RRD_FREE_SPACE_RESERVE_BYTES=2147483648
 uv run k1link serve
 ```
 
+Portable session products can additionally use the content-addressed central
+artifact gateway described in
+[`ADR 0033`](adr/0033-central-session-artifacts-and-bounded-host-cache.md).
+This does not move or delete the source-of-record evidence. It adds verified
+central manifests plus a quota-bound, pinnable local working set for exact base
+RRDs and perception overlays.
+
 The directories and derived recording cache use owner-only permissions where
 the host filesystem permits them. A new acquisition writer is assigned only a
 direct child of the private evidence root; it no longer writes a new run to the

docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md

@@ -0,0 +1,159 @@
+# ADR 0033: Central session artifacts and bounded operator-host cache
+
+Date: 2026-07-30
+Status: accepted and implemented
+
+## Decision
+
+Mission Core stores portable recorded-session products in a provider-neutral,
+content-addressed artifact store. The current central provider is the existing
+Synology SMB mount:
+
+```text
+/Volumes/docker/nodedc-mission-core/artifact-store
+```
+
+This is product runtime data. It is not a deploy artifact, a Map Gateway cache,
+a Device Plane package, or a new Docker service. Authentication remains owned
+by the operating-system SMB mount; Mission Core does not receive a NAS password.
+
+Every immutable object is stored under:
+
+```text
+objects/sha256/<first-two-hex>/<full-sha256>
+```
+
+Canonical JSON manifests are themselves addressed by SHA-256. Mutable names
+exist only as small, atomically replaced references:
+
+```text
+refs/<namespace>/<key>.json -> manifest SHA-256
+```
+
+An operator host never serves a central SMB file directly. It first streams the
+object into a temporary local file, proves the expected SHA-256 and byte length,
+and atomically publishes it into:
+
+```text
+<MISSIONCORE_DATA_DIR>/artifact-cache
+```
+
+The local cache has:
+
+- an 8 GiB default byte quota;
+- a 2 GiB default free-space reserve;
+- persistent SQLite metadata;
+- persistent pin sets;
+- LRU eviction of unpinned objects only;
+- no automatic deletion of source evidence;
+- offline resolution through locally sealed reference and manifest snapshots.
+
+The limits are configurable with:
+
+```text
+MISSIONCORE_ARTIFACT_CACHE_MAX_BYTES
+MISSIONCORE_ARTIFACT_CACHE_FREE_SPACE_RESERVE_BYTES
+```
+
+The central root is configured with:
+
+```text
+MISSIONCORE_ARTIFACT_STORE_ROOT
+```
+
+## Recorded playback behavior
+
+The base RRD and integrated E10 overlay are separate immutable roles in one
+recorded-session manifest.
+
+On a normal cache hit, recorded playback uses the existing private RRD cache.
+On a private-cache miss it may restore the exact `base-rrd` role from the
+bounded artifact cache. Restoration is admitted only when:
+
+- the manifest type and subject match the requested session;
+- the RRD content digest and length match the manifest;
+- the local source evidence still hashes to `base-source-sha256`;
+- the manifest carries the exact session timeline bounds.
+
+The restored RRD is republished through the existing private recording-cache
+transaction and receives a fresh sidecar bound to the current local evidence
+stats. The exporter is not run.
+
+The integrated-perception endpoint resolves
+`integrated-overlay:<recording-id>` before scanning or validating local worker
+results. A sealed central/local CAS hit therefore needs neither Worker 006 nor
+the original E10 result tree. Invalid central metadata fails closed; a genuinely
+absent central role can still use the existing local materialization path.
+
+## Persistent pins and offline use
+
+The CLI surface is:
+
+```bash
+k1link artifact publish ...
+k1link artifact pin ...
+k1link artifact unpin ...
+k1link artifact resolve ...
+k1link artifact status ...
+```
+
+Pinning is a local operator-host decision. A pin downloads and verifies every
+member of the selected manifest. If any member fails, the new pin is removed;
+objects already downloaded remain ordinary unpinned cache entries.
+
+When the central root is unavailable:
+
+- a sealed local reference, manifest, and object resolve normally;
+- an absent local object returns an explicit offline-cache error;
+- Mission Core does not silently fetch from Worker 006 or start a heavy AI run.
+
+## RAVNOVES00 migration acceptance
+
+The source session is `20260720T065719Z_viewer_live`, displayed as
+`RAVNOVES00`.
+
+The accepted central reference currently points to:
+
+```text
+092f758c09965be98fd76d8191474c528ee20540dce5f4761c2a087f6056a6d5
+```
+
+It contains nine logical roles totaling `721626746` bytes:
+
+- the source evidence archive;
+- the base RRD and its original audit sidecar;
+- the accepted E10 result archive;
+- the compute-job manifest;
+- the LiDAR pack and its manifest;
+- the sealed integrated overlay and its sidecar.
+
+The host pin is:
+
+```text
+session:20260720T065719Z_viewer_live
+```
+
+Acceptance facts:
+
+- all nine roles are cached and pinned;
+- base RRD SHA-256:
+  `4db5f2519caa61aca3683499094329b35cf3c24d1cc242953c54d2a228762e13`;
+- base source SHA-256:
+  `70da0edad5cbf0e89b6e2355c0aac294f8c36a67249cd33b8918353fd88c83af`;
+- overlay SHA-256:
+  `c2bb73fca8bb616a3929df236c6dc63d071b6466a6ca57d06487ef18ed985b57`;
+- an offline overlay resolve completed in 0.76 seconds;
+- an isolated offline base restore produced the exact RRD without invoking the
+  exporter;
+- an isolated offline overlay materialization succeeded with nonexistent local
+  worker-result, job, and LiDAR roots.
+
+The migration was copy-only. Existing evidence, derived recordings, accepted
+worker results, and historical overlay generations were not deleted.
+
+## Resource boundary
+
+Artifact publication, fetch, hashing, and restoration are sequential and use a
+1 MiB streaming buffer. They do not start Docker or parallel workers. This is
+the required operating envelope for the 14-inch 2023 MacBook Pro with 18 GiB
+RAM used as the current combined operator and test host.

src/k1link/artifact_cli.py

@@ -0,0 +1,248 @@
+"""Operator commands for the Mission Core central artifact gateway."""
+
+from __future__ import annotations
+
+import json
+import mimetypes
+import os
+from pathlib import Path
+from typing import Annotated
+
+import typer
+from rich.console import Console
+
+from k1link.artifact_gateway import (
+    CACHE_MAX_BYTES_ENV,
+    CACHE_RESERVE_BYTES_ENV,
+    DEFAULT_CACHE_MAX_BYTES,
+    DEFAULT_FREE_SPACE_RESERVE_BYTES,
+    STORE_ROOT_ENV,
+    ArtifactGateway,
+    ArtifactGatewayError,
+    CentralArtifactStore,
+    LocalArtifactCache,
+)
+from k1link.sessions.store import resolve_missioncore_data_dir
+
+app = typer.Typer(
+    help="Publish, pin, inspect, and resolve verified central Mission Core artifacts.",
+    no_args_is_help=True,
+)
+console = Console()
+
+
+def _positive_environment_bytes(name: str, default: int) -> int:
+    value = os.environ.get(name, "").strip()
+    if not value:
+        return default
+    if not value.isascii() or not value.isdecimal() or int(value) <= 0:
+        raise ValueError(f"{name} must be a positive integer")
+    return int(value)
+
+
+def _non_negative_environment_bytes(name: str, default: int) -> int:
+    value = os.environ.get(name, "").strip()
+    if not value:
+        return default
+    if not value.isascii() or not value.isdecimal():
+        raise ValueError(f"{name} must be a non-negative integer")
+    return int(value)
+
+
+def _gateway(repository_root: Path, *, create_store: bool = False) -> ArtifactGateway:
+    configured = os.environ.get(STORE_ROOT_ENV, "").strip()
+    if not configured:
+        raise ValueError(f"{STORE_ROOT_ENV} is not configured")
+    data_dir = resolve_missioncore_data_dir(repository_root)
+    return ArtifactGateway(
+        CentralArtifactStore(Path(configured), create=create_store),
+        LocalArtifactCache(
+            data_dir / "artifact-cache",
+            max_bytes=_positive_environment_bytes(
+                CACHE_MAX_BYTES_ENV,
+                DEFAULT_CACHE_MAX_BYTES,
+            ),
+            free_space_reserve_bytes=_non_negative_environment_bytes(
+                CACHE_RESERVE_BYTES_ENV,
+                DEFAULT_FREE_SPACE_RESERVE_BYTES,
+            ),
+        ),
+    )
+
+
+def _parse_assignment(value: str, *, label: str) -> tuple[str, str]:
+    key, separator, item = value.partition("=")
+    if not separator or not key or not item:
+        raise ValueError(f"{label} must use KEY=VALUE")
+    return key, item
+
+
+def _media_type(path: Path) -> str:
+    explicit = {
+        ".rrd": "application/vnd.rerun.rrd",
+        ".jsonl": "application/x-ndjson",
+        ".npz": "application/x-numpy",
+        ".k1mqtt": "application/vnd.nodedc.k1mqtt",
+        ".tar": "application/x-tar",
+        ".tgz": "application/gzip",
+        ".m4s": "video/iso.segment",
+    }.get(path.suffix.lower())
+    if explicit is not None:
+        return explicit
+    guessed, _encoding = mimetypes.guess_type(path.name)
+    return guessed or "application/octet-stream"
+
+
+@app.command("publish")
+def publish(
+    namespace: Annotated[str, typer.Option("--namespace")],
+    key: Annotated[str, typer.Option("--key")],
+    artifact_type: Annotated[str, typer.Option("--artifact-type")],
+    subject_id: Annotated[str, typer.Option("--subject-id")],
+    member: Annotated[
+        list[str],
+        typer.Option(
+            "--member",
+            help="Repeat ROLE=/absolute/file/path for every immutable bundle member.",
+        ),
+    ],
+    metadata: Annotated[
+        list[str] | None,
+        typer.Option("--metadata", help="Optional repeatable KEY=VALUE metadata."),
+    ] = None,
+) -> None:
+    """Publish exact files, then atomically advance one named central reference."""
+
+    repository_root = Path(__file__).resolve().parents[2]
+    try:
+        sources = []
+        for raw in member:
+            role, raw_path = _parse_assignment(raw, label="member")
+            path = Path(raw_path).expanduser().resolve(strict=True)
+            sources.append((role, _media_type(path), path))
+        metadata_values = dict(
+            _parse_assignment(raw, label="metadata") for raw in (metadata or [])
+        )
+        manifest = _gateway(repository_root, create_store=True).publish(
+            namespace=namespace,
+            key=key,
+            artifact_type=artifact_type,
+            subject_id=subject_id,
+            sources=sources,
+            metadata=metadata_values,
+        )
+    except (ArtifactGatewayError, OSError, ValueError) as exc:
+        console.print(f"[red]Artifact publication failed:[/red] {exc}")
+        raise typer.Exit(code=2) from exc
+    total_bytes = sum(item.byte_length for item in manifest.members)
+    console.print(
+        "[green]Central artifact published.[/green] "
+        f"reference={namespace}/{key}; manifest={manifest.manifest_id}; "
+        f"objects={len(manifest.members)}; logical_bytes={total_bytes}"
+    )
+
+
+@app.command("pin")
+def pin(
+    namespace: Annotated[str, typer.Option("--namespace")],
+    key: Annotated[str, typer.Option("--key")],
+    pin_id: Annotated[str | None, typer.Option("--pin-id")] = None,
+) -> None:
+    """Fetch and persistently pin every object in one named bundle."""
+
+    repository_root = Path(__file__).resolve().parents[2]
+    selected_pin = pin_id or f"{namespace}:{key}"
+    try:
+        gateway = _gateway(repository_root)
+        manifest = gateway.pin_reference(namespace, key, pin_id=selected_pin)
+        status = gateway.cache.status()
+    except (ArtifactGatewayError, OSError, ValueError) as exc:
+        console.print(f"[red]Artifact pin failed:[/red] {exc}")
+        raise typer.Exit(code=2) from exc
+    console.print(
+        "[green]Artifact bundle pinned.[/green] "
+        f"manifest={manifest.manifest_id}; pin={selected_pin}; "
+        f"pinned_objects={status.pinned_object_count}; pinned_bytes={status.pinned_bytes}"
+    )
+
+
+@app.command("unpin")
+def unpin(pin_id: Annotated[str, typer.Option("--pin-id")]) -> None:
+    """Remove one persistent pin and immediately enforce local LRU limits."""
+
+    repository_root = Path(__file__).resolve().parents[2]
+    try:
+        gateway = _gateway(repository_root)
+        gateway.cache.unpin(pin_id)
+        status = gateway.cache.status()
+    except (ArtifactGatewayError, OSError, ValueError) as exc:
+        console.print(f"[red]Artifact unpin failed:[/red] {exc}")
+        raise typer.Exit(code=2) from exc
+    console.print(
+        "[green]Artifact pin removed.[/green] "
+        f"pin={pin_id}; cached_bytes={status.total_bytes}; "
+        f"pinned_bytes={status.pinned_bytes}"
+    )
+
+
+@app.command("resolve")
+def resolve(
+    namespace: Annotated[str, typer.Option("--namespace")],
+    key: Annotated[str, typer.Option("--key")],
+    role: Annotated[str, typer.Option("--role")],
+    json_output: Annotated[bool, typer.Option("--json")] = False,
+) -> None:
+    """Resolve one role to a verified local path, fetching it when online."""
+
+    repository_root = Path(__file__).resolve().parents[2]
+    try:
+        artifact = _gateway(repository_root).resolve_role(namespace, key, role)
+    except (ArtifactGatewayError, OSError, ValueError) as exc:
+        console.print(f"[red]Artifact resolution failed:[/red] {exc}")
+        raise typer.Exit(code=2) from exc
+    document = {
+        "manifest_id": artifact.manifest.manifest_id,
+        "role": artifact.member.role,
+        "sha256": artifact.member.sha256,
+        "byte_length": artifact.member.byte_length,
+        "path": str(artifact.path),
+        "cache_hit": artifact.cache_hit,
+        "central_available": artifact.central_available,
+    }
+    if json_output:
+        typer.echo(json.dumps(document, ensure_ascii=False, indent=2))
+        return
+    console.print(
+        "[green]Artifact resolved.[/green] "
+        f"role={role}; path={artifact.path}; cache_hit={artifact.cache_hit}; "
+        f"central_available={artifact.central_available}"
+    )
+
+
+@app.command("status")
+def status(json_output: Annotated[bool, typer.Option("--json")] = False) -> None:
+    """Report local working-set quota and persistent pin usage."""
+
+    repository_root = Path(__file__).resolve().parents[2]
+    try:
+        cache_status = _gateway(repository_root).cache.status()
+    except (ArtifactGatewayError, OSError, ValueError) as exc:
+        console.print(f"[red]Artifact status failed:[/red] {exc}")
+        raise typer.Exit(code=2) from exc
+    document = {
+        "object_count": cache_status.object_count,
+        "total_bytes": cache_status.total_bytes,
+        "pinned_object_count": cache_status.pinned_object_count,
+        "pinned_bytes": cache_status.pinned_bytes,
+        "cache_max_bytes": cache_status.cache_max_bytes,
+        "free_space_reserve_bytes": cache_status.free_space_reserve_bytes,
+    }
+    if json_output:
+        typer.echo(json.dumps(document, ensure_ascii=False, indent=2))
+        return
+    console.print(
+        f"objects={cache_status.object_count}; bytes={cache_status.total_bytes}; "
+        f"pinned_objects={cache_status.pinned_object_count}; "
+        f"pinned_bytes={cache_status.pinned_bytes}; "
+        f"quota={cache_status.cache_max_bytes}"
+    )

src/k1link/artifact_gateway.py

@@ -0,0 +1,954 @@
+"""Content-addressed central artifacts with a bounded, offline-capable host cache."""
+
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import re
+import shutil
+import sqlite3
+import stat
+import threading
+import time
+from collections.abc import Iterator, Mapping, Sequence
+from contextlib import contextmanager, suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Final
+from uuid import uuid4
+
+from k1link.artifacts import utc_now_iso
+
+MANIFEST_SCHEMA: Final = "missioncore.artifact-manifest/v1"
+REFERENCE_SCHEMA: Final = "missioncore.artifact-reference/v1"
+DEFAULT_CACHE_MAX_BYTES: Final = 8 * 1024 * 1024 * 1024
+DEFAULT_FREE_SPACE_RESERVE_BYTES: Final = 2 * 1024 * 1024 * 1024
+STORE_ROOT_ENV: Final = "MISSIONCORE_ARTIFACT_STORE_ROOT"
+CACHE_MAX_BYTES_ENV: Final = "MISSIONCORE_ARTIFACT_CACHE_MAX_BYTES"
+CACHE_RESERVE_BYTES_ENV: Final = "MISSIONCORE_ARTIFACT_CACHE_FREE_SPACE_RESERVE_BYTES"
+
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+_SAFE_COMPONENT = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
+_MAX_MANIFEST_BYTES = 8 * 1024 * 1024
+_COPY_CHUNK_BYTES = 1024 * 1024
+
+
+class ArtifactGatewayError(RuntimeError):
+    """Base class for central artifact and host-cache failures."""
+
+
+class ArtifactStoreUnavailable(ArtifactGatewayError):
+    """The configured central store cannot currently be reached."""
+
+
+class ArtifactNotFound(ArtifactGatewayError):
+    """An exact reference, manifest, role, or object does not exist."""
+
+
+class ArtifactIntegrityError(ArtifactGatewayError):
+    """An artifact violates its content-addressed identity."""
+
+
+class ArtifactCacheCapacityError(ArtifactGatewayError):
+    """The local host cannot safely reserve enough cache capacity."""
+
+
+@dataclass(frozen=True, slots=True)
+class ArtifactMember:
+    role: str
+    media_type: str
+    sha256: str
+    byte_length: int
+
+
+@dataclass(frozen=True, slots=True)
+class ArtifactManifest:
+    manifest_id: str
+    artifact_type: str
+    subject_id: str
+    created_at_utc: str
+    members: tuple[ArtifactMember, ...]
+    metadata: Mapping[str, str]
+
+    def member(self, role: str) -> ArtifactMember:
+        matches = tuple(item for item in self.members if item.role == role)
+        if len(matches) != 1:
+            raise ArtifactNotFound(f"artifact role is unavailable: {role}")
+        return matches[0]
+
+
+@dataclass(frozen=True, slots=True)
+class ResolvedArtifact:
+    manifest: ArtifactManifest
+    member: ArtifactMember
+    path: Path
+    cache_hit: bool
+    central_available: bool
+
+
+@dataclass(frozen=True, slots=True)
+class ArtifactCacheStatus:
+    object_count: int
+    total_bytes: int
+    pinned_object_count: int
+    pinned_bytes: int
+    cache_max_bytes: int
+    free_space_reserve_bytes: int
+
+
+class CentralArtifactStore:
+    """Immutable SHA-256 objects/manifests plus atomically replaceable named refs."""
+
+    def __init__(self, root: Path, *, create: bool = False) -> None:
+        self.root = root.expanduser().absolute()
+        if create:
+            try:
+                self.root.mkdir(parents=True, exist_ok=True)
+            except OSError as exc:
+                raise ArtifactStoreUnavailable("central artifact store cannot be created") from exc
+            self._require_root()
+
+    def publish_file(self, source: Path) -> ArtifactMember:
+        source_path = _regular_source(source)
+        digest, byte_length = _hash_file(source_path)
+        destination = self.object_path(digest)
+        try:
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            if destination.exists() or destination.is_symlink():
+                _verify_object(destination, digest, byte_length)
+            else:
+                temporary = destination.with_name(f".{digest}.{uuid4().hex}.tmp")
+                try:
+                    copied_digest, copied_bytes = _copy_and_hash(source_path, temporary)
+                    if copied_digest != digest or copied_bytes != byte_length:
+                        raise ArtifactIntegrityError("source changed while it was published")
+                    os.replace(temporary, destination)
+                finally:
+                    temporary.unlink(missing_ok=True)
+        except ArtifactGatewayError:
+            raise
+        except OSError as exc:
+            raise ArtifactStoreUnavailable("central artifact object cannot be published") from exc
+        return ArtifactMember(
+            role="",
+            media_type="application/octet-stream",
+            sha256=digest,
+            byte_length=byte_length,
+        )
+
+    def publish_manifest(
+        self,
+        *,
+        artifact_type: str,
+        subject_id: str,
+        members: Sequence[ArtifactMember],
+        metadata: Mapping[str, str] | None = None,
+        created_at_utc: str | None = None,
+    ) -> ArtifactManifest:
+        _validate_component(artifact_type, "artifact type")
+        _validate_component(subject_id, "artifact subject")
+        normalized_members = tuple(sorted(members, key=lambda item: item.role))
+        _validate_members(normalized_members)
+        normalized_metadata = dict(sorted((metadata or {}).items()))
+        for key, value in normalized_metadata.items():
+            _validate_component(key, "metadata key")
+            if not isinstance(value, str) or len(value) > 4096:
+                raise ValueError("artifact metadata value is invalid")
+        document = {
+            "schema_version": MANIFEST_SCHEMA,
+            "artifact_type": artifact_type,
+            "subject_id": subject_id,
+            "created_at_utc": created_at_utc or utc_now_iso(),
+            "members": [
+                {
+                    "role": item.role,
+                    "media_type": item.media_type,
+                    "sha256": item.sha256,
+                    "byte_length": item.byte_length,
+                }
+                for item in normalized_members
+            ],
+            "metadata": normalized_metadata,
+        }
+        payload = _canonical_json(document)
+        manifest_id = hashlib.sha256(payload).hexdigest()
+        destination = self.manifest_path(manifest_id)
+        try:
+            destination.parent.mkdir(parents=True, exist_ok=True)
+            if destination.exists() or destination.is_symlink():
+                if destination.read_bytes() != payload:
+                    raise ArtifactIntegrityError("central artifact manifest digest is inconsistent")
+            else:
+                _write_bytes_atomic(destination, payload)
+        except ArtifactGatewayError:
+            raise
+        except OSError as exc:
+            raise ArtifactStoreUnavailable("central artifact manifest cannot be published") from exc
+        return _parse_manifest(document, manifest_id)
+
+    def set_reference(self, namespace: str, key: str, manifest_id: str) -> None:
+        _validate_component(namespace, "artifact namespace")
+        _validate_component(key, "artifact reference key")
+        _validate_sha256(manifest_id)
+        self.read_manifest(manifest_id)
+        document = {
+            "schema_version": REFERENCE_SCHEMA,
+            "namespace": namespace,
+            "key": key,
+            "manifest_id": manifest_id,
+            "updated_at_utc": utc_now_iso(),
+        }
+        try:
+            _write_bytes_atomic(self.reference_path(namespace, key), _canonical_json(document))
+        except OSError as exc:
+            raise ArtifactStoreUnavailable(
+                "central artifact reference cannot be published"
+            ) from exc
+
+    def resolve_reference(self, namespace: str, key: str) -> ArtifactManifest:
+        reference = self.read_reference_document(namespace, key)
+        return self.read_manifest(str(reference["manifest_id"]))
+
+    def read_reference_document(self, namespace: str, key: str) -> dict[str, Any]:
+        _validate_component(namespace, "artifact namespace")
+        _validate_component(key, "artifact reference key")
+        path = self.reference_path(namespace, key)
+        document = _read_json_document(path, unavailable_message="central reference is unavailable")
+        if (
+            document.get("schema_version") != REFERENCE_SCHEMA
+            or document.get("namespace") != namespace
+            or document.get("key") != key
+            or not isinstance(document.get("manifest_id"), str)
+            or _SHA256.fullmatch(str(document["manifest_id"])) is None
+            or not isinstance(document.get("updated_at_utc"), str)
+        ):
+            raise ArtifactIntegrityError("central artifact reference is invalid")
+        return document
+
+    def read_manifest(self, manifest_id: str) -> ArtifactManifest:
+        _validate_sha256(manifest_id)
+        path = self.manifest_path(manifest_id)
+        document = _read_json_document(path, unavailable_message="central manifest is unavailable")
+        payload = _canonical_json(document)
+        if hashlib.sha256(payload).hexdigest() != manifest_id:
+            raise ArtifactIntegrityError("central artifact manifest digest changed")
+        return _parse_manifest(document, manifest_id)
+
+    def object_path(self, digest: str) -> Path:
+        _validate_sha256(digest)
+        return self.root / "objects" / "sha256" / digest[:2] / digest
+
+    def manifest_path(self, manifest_id: str) -> Path:
+        _validate_sha256(manifest_id)
+        return self.root / "manifests" / "sha256" / manifest_id[:2] / f"{manifest_id}.json"
+
+    def reference_path(self, namespace: str, key: str) -> Path:
+        _validate_component(namespace, "artifact namespace")
+        _validate_component(key, "artifact reference key")
+        return self.root / "refs" / namespace / f"{key}.json"
+
+    def _require_root(self) -> None:
+        try:
+            metadata = self.root.lstat()
+        except OSError as exc:
+            raise ArtifactStoreUnavailable("central artifact store is unavailable") from exc
+        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISDIR(metadata.st_mode):
+            raise ArtifactIntegrityError("central artifact store root is invalid")
+
+
+class LocalArtifactCache:
+    """Persistent local CAS with durable pins and unpinned LRU eviction."""
+
+    def __init__(
+        self,
+        root: Path,
+        *,
+        max_bytes: int = DEFAULT_CACHE_MAX_BYTES,
+        free_space_reserve_bytes: int = DEFAULT_FREE_SPACE_RESERVE_BYTES,
+    ) -> None:
+        if max_bytes <= 0:
+            raise ValueError("artifact cache maximum bytes must be positive")
+        if free_space_reserve_bytes < 0:
+            raise ValueError("artifact cache free-space reserve must be non-negative")
+        self.root = root.expanduser().absolute()
+        self.root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        root_metadata = self.root.lstat()
+        if stat.S_ISLNK(root_metadata.st_mode) or not stat.S_ISDIR(root_metadata.st_mode):
+            raise ArtifactIntegrityError("local artifact cache root is invalid")
+        with suppress(OSError):
+            self.root.chmod(0o700)
+        self.objects_root = self.root / "objects" / "sha256"
+        self.metadata_root = self.root / "metadata"
+        self.objects_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        self.metadata_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        self.database_path = self.root / "cache.sqlite3"
+        self.lock_path = self.root / ".cache.lock"
+        self.max_bytes = max_bytes
+        self.free_space_reserve_bytes = free_space_reserve_bytes
+        self._thread_lock = threading.RLock()
+        self._initialize_database()
+
+    def get(self, member: ArtifactMember) -> Path | None:
+        with self._locked():
+            return self._get_locked(member, touch=True)
+
+    def fetch(
+        self,
+        store: CentralArtifactStore,
+        member: ArtifactMember,
+        *,
+        pin_id: str | None = None,
+    ) -> tuple[Path, bool]:
+        _validate_member(member)
+        if pin_id is not None:
+            _validate_component(pin_id, "artifact pin id")
+        with self._locked():
+            cached = self._get_locked(member, touch=True)
+            if cached is not None:
+                if pin_id is not None:
+                    self._pin_locked(pin_id, member.sha256)
+                return cached, True
+            self._reserve_locked(member.byte_length, protected_digest=member.sha256)
+            source = store.object_path(member.sha256)
+            destination = self.object_path(member.sha256)
+            destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            temporary = destination.with_name(f".{member.sha256}.{uuid4().hex}.tmp")
+            try:
+                copied_digest, copied_bytes = _copy_and_hash(source, temporary)
+                if copied_digest != member.sha256 or copied_bytes != member.byte_length:
+                    raise ArtifactIntegrityError("central artifact object failed verification")
+                os.chmod(temporary, 0o600)
+                os.replace(temporary, destination)
+            except ArtifactGatewayError:
+                raise
+            except OSError as exc:
+                raise ArtifactStoreUnavailable("central artifact object is unavailable") from exc
+            finally:
+                temporary.unlink(missing_ok=True)
+            metadata = destination.stat()
+            with self._connect() as connection:
+                connection.execute(
+                    """
+                    INSERT INTO objects(
+                        sha256, byte_length, last_access_ns, mtime_ns, inode
+                    ) VALUES (?, ?, ?, ?, ?)
+                    ON CONFLICT(sha256) DO UPDATE SET
+                        byte_length=excluded.byte_length,
+                        last_access_ns=excluded.last_access_ns,
+                        mtime_ns=excluded.mtime_ns,
+                        inode=excluded.inode
+                    """,
+                    (
+                        member.sha256,
+                        member.byte_length,
+                        time.time_ns(),
+                        metadata.st_mtime_ns,
+                        metadata.st_ino,
+                    ),
+                )
+                if pin_id is not None:
+                    connection.execute(
+                        "INSERT OR IGNORE INTO pins(pin_id, sha256) VALUES (?, ?)",
+                        (pin_id, member.sha256),
+                    )
+            self._evict_locked(protected_digest=member.sha256)
+            return destination.resolve(strict=True), False
+
+    def pin(self, pin_id: str, members: Sequence[ArtifactMember]) -> tuple[Path, ...]:
+        _validate_component(pin_id, "artifact pin id")
+        paths: list[Path] = []
+        try:
+            for member in members:
+                path = self.get(member)
+                if path is None:
+                    raise ArtifactNotFound(
+                        f"artifact object must be fetched before pinning: {member.role}"
+                    )
+                with self._locked():
+                    self._pin_locked(pin_id, member.sha256)
+                paths.append(path)
+        except BaseException:
+            self.unpin(pin_id)
+            raise
+        return tuple(paths)
+
+    def unpin(self, pin_id: str) -> None:
+        _validate_component(pin_id, "artifact pin id")
+        with self._locked(), self._connect() as connection:
+            connection.execute("DELETE FROM pins WHERE pin_id = ?", (pin_id,))
+        with self._locked():
+            self._evict_locked(protected_digest=None)
+
+    def save_snapshot(
+        self,
+        namespace: str,
+        key: str,
+        reference: Mapping[str, Any],
+        manifest: ArtifactManifest,
+    ) -> None:
+        _validate_component(namespace, "artifact namespace")
+        _validate_component(key, "artifact reference key")
+        manifest_document = _manifest_document(manifest)
+        reference_document = dict(reference)
+        _validate_reference_snapshot(reference_document, namespace, key)
+        if reference_document["manifest_id"] != manifest.manifest_id:
+            raise ArtifactIntegrityError("artifact reference and manifest do not match")
+        manifest_path = (
+            self.metadata_root
+            / "manifests"
+            / manifest.manifest_id[:2]
+            / f"{manifest.manifest_id}.json"
+        )
+        reference_path = self.metadata_root / "refs" / namespace / f"{key}.json"
+        with self._locked():
+            _write_bytes_atomic(manifest_path, _canonical_json(manifest_document), mode=0o600)
+            _write_bytes_atomic(reference_path, _canonical_json(reference_document), mode=0o600)
+
+    def load_snapshot(self, namespace: str, key: str) -> ArtifactManifest:
+        _validate_component(namespace, "artifact namespace")
+        _validate_component(key, "artifact reference key")
+        reference_path = self.metadata_root / "refs" / namespace / f"{key}.json"
+        try:
+            reference = _read_json_document(
+                reference_path,
+                unavailable_message="local artifact reference is unavailable",
+            )
+        except ArtifactStoreUnavailable as exc:
+            raise ArtifactNotFound("local artifact reference is unavailable") from exc
+        _validate_reference_snapshot(reference, namespace, key)
+        manifest_id = str(reference["manifest_id"])
+        manifest_path = (
+            self.metadata_root / "manifests" / manifest_id[:2] / f"{manifest_id}.json"
+        )
+        try:
+            document = _read_json_document(
+                manifest_path,
+                unavailable_message="local artifact manifest is unavailable",
+            )
+        except ArtifactStoreUnavailable as exc:
+            raise ArtifactNotFound("local artifact manifest is unavailable") from exc
+        if hashlib.sha256(_canonical_json(document)).hexdigest() != manifest_id:
+            raise ArtifactIntegrityError("local artifact manifest digest changed")
+        return _parse_manifest(document, manifest_id)
+
+    def status(self) -> ArtifactCacheStatus:
+        with self._locked(), self._connect() as connection:
+            row = connection.execute(
+                """
+                SELECT
+                    COUNT(*),
+                    COALESCE(SUM(byte_length), 0),
+                    COUNT(DISTINCT pins.sha256),
+                    COALESCE(SUM(
+                        CASE WHEN pins.sha256 IS NOT NULL THEN objects.byte_length ELSE 0 END
+                    ), 0)
+                FROM objects
+                LEFT JOIN (SELECT DISTINCT sha256 FROM pins) AS pins USING (sha256)
+                """
+            ).fetchone()
+        assert row is not None
+        return ArtifactCacheStatus(
+            object_count=int(row[0]),
+            total_bytes=int(row[1]),
+            pinned_object_count=int(row[2]),
+            pinned_bytes=int(row[3]),
+            cache_max_bytes=self.max_bytes,
+            free_space_reserve_bytes=self.free_space_reserve_bytes,
+        )
+
+    def object_path(self, digest: str) -> Path:
+        _validate_sha256(digest)
+        return self.objects_root / digest[:2] / digest
+
+    def _initialize_database(self) -> None:
+        with self._connect() as connection:
+            connection.executescript(
+                """
+                PRAGMA journal_mode = WAL;
+                PRAGMA foreign_keys = ON;
+                CREATE TABLE IF NOT EXISTS objects (
+                    sha256 TEXT PRIMARY KEY,
+                    byte_length INTEGER NOT NULL CHECK (byte_length >= 0),
+                    last_access_ns INTEGER NOT NULL,
+                    mtime_ns INTEGER NOT NULL,
+                    inode INTEGER NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS pins (
+                    pin_id TEXT NOT NULL,
+                    sha256 TEXT NOT NULL REFERENCES objects(sha256) ON DELETE CASCADE,
+                    PRIMARY KEY(pin_id, sha256)
+                );
+                CREATE INDEX IF NOT EXISTS objects_lru ON objects(last_access_ns, sha256);
+                CREATE INDEX IF NOT EXISTS pins_digest ON pins(sha256);
+                """
+            )
+        with suppress(OSError):
+            self.database_path.chmod(0o600)
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.database_path, timeout=30.0)
+        connection.execute("PRAGMA foreign_keys = ON")
+        return connection
+
+    @contextmanager
+    def _locked(self) -> Iterator[None]:
+        with self._thread_lock:
+            descriptor = os.open(self.lock_path, os.O_CREAT | os.O_RDWR, 0o600)
+            try:
+                fcntl.flock(descriptor, fcntl.LOCK_EX)
+                yield
+            finally:
+                fcntl.flock(descriptor, fcntl.LOCK_UN)
+                os.close(descriptor)
+
+    def _get_locked(self, member: ArtifactMember, *, touch: bool) -> Path | None:
+        _validate_member(member)
+        path = self.object_path(member.sha256)
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT byte_length, mtime_ns, inode FROM objects WHERE sha256 = ?",
+                (member.sha256,),
+            ).fetchone()
+            if row is None:
+                return None
+            try:
+                metadata = path.lstat()
+            except OSError:
+                connection.execute("DELETE FROM objects WHERE sha256 = ?", (member.sha256,))
+                return None
+            if (
+                stat.S_ISLNK(metadata.st_mode)
+                or not stat.S_ISREG(metadata.st_mode)
+                or metadata.st_size != member.byte_length
+                or int(row[0]) != member.byte_length
+            ):
+                path.unlink(missing_ok=True)
+                connection.execute("DELETE FROM objects WHERE sha256 = ?", (member.sha256,))
+                return None
+            if metadata.st_mtime_ns != int(row[1]) or metadata.st_ino != int(row[2]):
+                digest, byte_length = _hash_file(path)
+                if digest != member.sha256 or byte_length != member.byte_length:
+                    path.unlink(missing_ok=True)
+                    connection.execute("DELETE FROM objects WHERE sha256 = ?", (member.sha256,))
+                    return None
+                connection.execute(
+                    "UPDATE objects SET mtime_ns = ?, inode = ? WHERE sha256 = ?",
+                    (metadata.st_mtime_ns, metadata.st_ino, member.sha256),
+                )
+            if touch:
+                connection.execute(
+                    "UPDATE objects SET last_access_ns = ? WHERE sha256 = ?",
+                    (time.time_ns(), member.sha256),
+                )
+        return path.resolve(strict=True)
+
+    def _pin_locked(self, pin_id: str, digest: str) -> None:
+        with self._connect() as connection:
+            connection.execute(
+                "INSERT OR IGNORE INTO pins(pin_id, sha256) VALUES (?, ?)",
+                (pin_id, digest),
+            )
+
+    def _reserve_locked(self, required_bytes: int, *, protected_digest: str) -> None:
+        if required_bytes < 0:
+            raise ArtifactCacheCapacityError("artifact cache reservation is invalid")
+        if required_bytes > self.max_bytes:
+            raise ArtifactCacheCapacityError("artifact exceeds the local cache quota")
+        self._evict_locked(
+            protected_digest=protected_digest,
+            incoming_bytes=required_bytes,
+        )
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT COALESCE(SUM(byte_length), 0) FROM objects"
+            ).fetchone()
+        total_bytes = int(row[0]) if row is not None else 0
+        free_bytes = shutil.disk_usage(self.root).free
+        if (
+            total_bytes + required_bytes > self.max_bytes
+            or free_bytes - required_bytes < self.free_space_reserve_bytes
+        ):
+            raise ArtifactCacheCapacityError(
+                "local artifact cache cannot preserve its quota and free-space reserve"
+            )
+
+    def _evict_locked(
+        self,
+        *,
+        protected_digest: str | None,
+        incoming_bytes: int = 0,
+    ) -> None:
+        while True:
+            with self._connect() as connection:
+                total_row = connection.execute(
+                    "SELECT COALESCE(SUM(byte_length), 0) FROM objects"
+                ).fetchone()
+                total_bytes = int(total_row[0]) if total_row is not None else 0
+                free_bytes = shutil.disk_usage(self.root).free
+                if (
+                    total_bytes + incoming_bytes <= self.max_bytes
+                    and free_bytes - incoming_bytes >= self.free_space_reserve_bytes
+                ):
+                    return
+                row = connection.execute(
+                    """
+                    SELECT objects.sha256
+                    FROM objects
+                    LEFT JOIN pins ON pins.sha256 = objects.sha256
+                    WHERE pins.sha256 IS NULL AND objects.sha256 != COALESCE(?, '')
+                    ORDER BY objects.last_access_ns ASC, objects.sha256 ASC
+                    LIMIT 1
+                    """,
+                    (protected_digest,),
+                ).fetchone()
+                if row is None:
+                    return
+                digest = str(row[0])
+                self.object_path(digest).unlink(missing_ok=True)
+                connection.execute("DELETE FROM objects WHERE sha256 = ?", (digest,))
+
+
+class ArtifactGateway:
+    """Resolve named central manifests through a verified local working set."""
+
+    def __init__(self, store: CentralArtifactStore, cache: LocalArtifactCache) -> None:
+        self.store = store
+        self.cache = cache
+
+    def publish(
+        self,
+        *,
+        namespace: str,
+        key: str,
+        artifact_type: str,
+        subject_id: str,
+        sources: Sequence[tuple[str, str, Path]],
+        metadata: Mapping[str, str] | None = None,
+    ) -> ArtifactManifest:
+        members: list[ArtifactMember] = []
+        seen_roles: set[str] = set()
+        for role, media_type, source in sources:
+            _validate_component(role, "artifact role")
+            if role in seen_roles:
+                raise ValueError(f"duplicate artifact role: {role}")
+            seen_roles.add(role)
+            published = self.store.publish_file(source)
+            members.append(
+                ArtifactMember(
+                    role=role,
+                    media_type=media_type,
+                    sha256=published.sha256,
+                    byte_length=published.byte_length,
+                )
+            )
+        manifest = self.store.publish_manifest(
+            artifact_type=artifact_type,
+            subject_id=subject_id,
+            members=members,
+            metadata=metadata,
+        )
+        self.store.set_reference(namespace, key, manifest.manifest_id)
+        reference = self.store.read_reference_document(namespace, key)
+        self.cache.save_snapshot(namespace, key, reference, manifest)
+        return manifest
+
+    def resolve_role(self, namespace: str, key: str, role: str) -> ResolvedArtifact:
+        manifest, central_available = self._resolve_manifest(namespace, key)
+        member = manifest.member(role)
+        cached = self.cache.get(member)
+        if cached is not None:
+            return ResolvedArtifact(
+                manifest=manifest,
+                member=member,
+                path=cached,
+                cache_hit=True,
+                central_available=central_available,
+            )
+        if not central_available:
+            raise ArtifactStoreUnavailable(
+                f"artifact role is not in the offline cache: {namespace}/{key}/{role}"
+            )
+        path, cache_hit = self.cache.fetch(self.store, member)
+        return ResolvedArtifact(
+            manifest=manifest,
+            member=member,
+            path=path,
+            cache_hit=cache_hit,
+            central_available=True,
+        )
+
+    def pin_reference(self, namespace: str, key: str, *, pin_id: str) -> ArtifactManifest:
+        manifest, central_available = self._resolve_manifest(namespace, key)
+        fetched: list[ArtifactMember] = []
+        try:
+            for member in manifest.members:
+                if self.cache.get(member) is None:
+                    if not central_available:
+                        raise ArtifactStoreUnavailable(
+                            f"artifact role is not in the offline cache: {member.role}"
+                        )
+                    self.cache.fetch(self.store, member, pin_id=pin_id)
+                else:
+                    with self.cache._locked():
+                        self.cache._pin_locked(pin_id, member.sha256)
+                fetched.append(member)
+        except BaseException:
+            self.cache.unpin(pin_id)
+            raise
+        return manifest
+
+    def _resolve_manifest(
+        self,
+        namespace: str,
+        key: str,
+    ) -> tuple[ArtifactManifest, bool]:
+        try:
+            reference = self.store.read_reference_document(namespace, key)
+            manifest = self.store.read_manifest(str(reference["manifest_id"]))
+        except ArtifactStoreUnavailable:
+            return self.cache.load_snapshot(namespace, key), False
+        self.cache.save_snapshot(namespace, key, reference, manifest)
+        return manifest, True
+
+
+def configured_artifact_gateway(data_dir: Path) -> ArtifactGateway | None:
+    """Build the optional production gateway from environment-only configuration."""
+
+    configured_root = os.environ.get(STORE_ROOT_ENV, "").strip()
+    if not configured_root:
+        return None
+    max_bytes = _configured_non_negative_int(CACHE_MAX_BYTES_ENV, DEFAULT_CACHE_MAX_BYTES)
+    reserve_bytes = _configured_non_negative_int(
+        CACHE_RESERVE_BYTES_ENV,
+        DEFAULT_FREE_SPACE_RESERVE_BYTES,
+    )
+    if max_bytes <= 0:
+        raise ArtifactCacheCapacityError("artifact cache maximum bytes must be positive")
+    return ArtifactGateway(
+        CentralArtifactStore(Path(configured_root), create=False),
+        LocalArtifactCache(
+            data_dir / "artifact-cache",
+            max_bytes=max_bytes,
+            free_space_reserve_bytes=reserve_bytes,
+        ),
+    )
+
+
+def _configured_non_negative_int(name: str, default: int) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    if not raw.isascii() or not raw.isdecimal():
+        raise ValueError(f"{name} must be a non-negative integer")
+    return int(raw)
+
+
+def _regular_source(path: Path) -> Path:
+    try:
+        metadata = path.lstat()
+        resolved = path.expanduser().resolve(strict=True)
+    except OSError as exc:
+        raise ArtifactNotFound("artifact source is unavailable") from exc
+    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
+        raise ArtifactIntegrityError("artifact source must be a regular file")
+    return resolved
+
+
+def _hash_file(path: Path) -> tuple[str, int]:
+    digest = hashlib.sha256()
+    byte_length = 0
+    try:
+        with path.open("rb") as stream:
+            while chunk := stream.read(_COPY_CHUNK_BYTES):
+                digest.update(chunk)
+                byte_length += len(chunk)
+    except OSError as exc:
+        raise ArtifactStoreUnavailable("artifact object is unavailable") from exc
+    return digest.hexdigest(), byte_length
+
+
+def _copy_and_hash(source: Path, destination: Path) -> tuple[str, int]:
+    digest = hashlib.sha256()
+    byte_length = 0
+    with source.open("rb") as input_stream, destination.open("xb") as output_stream:
+        while chunk := input_stream.read(_COPY_CHUNK_BYTES):
+            output_stream.write(chunk)
+            digest.update(chunk)
+            byte_length += len(chunk)
+        output_stream.flush()
+        os.fsync(output_stream.fileno())
+    return digest.hexdigest(), byte_length
+
+
+def _verify_object(path: Path, digest: str, byte_length: int) -> None:
+    metadata = path.lstat()
+    if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
+        raise ArtifactIntegrityError("content-addressed object path is invalid")
+    actual_digest, actual_bytes = _hash_file(path)
+    if actual_digest != digest or actual_bytes != byte_length:
+        raise ArtifactIntegrityError("content-addressed object is corrupt")
+
+
+def _validate_members(members: Sequence[ArtifactMember]) -> None:
+    if not members:
+        raise ValueError("artifact manifest must contain at least one member")
+    roles: set[str] = set()
+    for member in members:
+        _validate_member(member)
+        if member.role in roles:
+            raise ValueError(f"duplicate artifact role: {member.role}")
+        roles.add(member.role)
+
+
+def _validate_member(member: ArtifactMember) -> None:
+    _validate_component(member.role, "artifact role")
+    if not isinstance(member.media_type, str) or not 1 <= len(member.media_type) <= 255:
+        raise ValueError("artifact media type is invalid")
+    _validate_sha256(member.sha256)
+    if not isinstance(member.byte_length, int) or isinstance(member.byte_length, bool):
+        raise ValueError("artifact byte length is invalid")
+    if member.byte_length < 0:
+        raise ValueError("artifact byte length is invalid")
+
+
+def _parse_manifest(document: Mapping[str, Any], manifest_id: str) -> ArtifactManifest:
+    if (
+        document.get("schema_version") != MANIFEST_SCHEMA
+        or not isinstance(document.get("artifact_type"), str)
+        or not isinstance(document.get("subject_id"), str)
+        or not isinstance(document.get("created_at_utc"), str)
+        or not isinstance(document.get("members"), list)
+        or not isinstance(document.get("metadata"), dict)
+    ):
+        raise ArtifactIntegrityError("artifact manifest shape is invalid")
+    _validate_component(str(document["artifact_type"]), "artifact type")
+    _validate_component(str(document["subject_id"]), "artifact subject")
+    raw_metadata = document["metadata"]
+    assert isinstance(raw_metadata, dict)
+    metadata: dict[str, str] = {}
+    for key, value in raw_metadata.items():
+        if not isinstance(key, str) or not isinstance(value, str):
+            raise ArtifactIntegrityError("artifact manifest metadata is invalid")
+        _validate_component(key, "metadata key")
+        metadata[key] = value
+    raw_members = document["members"]
+    assert isinstance(raw_members, list)
+    members: list[ArtifactMember] = []
+    for raw in raw_members:
+        if not isinstance(raw, dict):
+            raise ArtifactIntegrityError("artifact manifest member is invalid")
+        try:
+            member = ArtifactMember(
+                role=str(raw["role"]),
+                media_type=str(raw["media_type"]),
+                sha256=str(raw["sha256"]),
+                byte_length=int(raw["byte_length"]),
+            )
+        except (KeyError, TypeError, ValueError) as exc:
+            raise ArtifactIntegrityError("artifact manifest member is invalid") from exc
+        _validate_member(member)
+        members.append(member)
+    _validate_members(members)
+    if tuple(members) != tuple(sorted(members, key=lambda item: item.role)):
+        raise ArtifactIntegrityError("artifact manifest members are not canonical")
+    return ArtifactManifest(
+        manifest_id=manifest_id,
+        artifact_type=str(document["artifact_type"]),
+        subject_id=str(document["subject_id"]),
+        created_at_utc=str(document["created_at_utc"]),
+        members=tuple(members),
+        metadata=metadata,
+    )
+
+
+def _manifest_document(manifest: ArtifactManifest) -> dict[str, Any]:
+    return {
+        "schema_version": MANIFEST_SCHEMA,
+        "artifact_type": manifest.artifact_type,
+        "subject_id": manifest.subject_id,
+        "created_at_utc": manifest.created_at_utc,
+        "members": [
+            {
+                "role": item.role,
+                "media_type": item.media_type,
+                "sha256": item.sha256,
+                "byte_length": item.byte_length,
+            }
+            for item in manifest.members
+        ],
+        "metadata": dict(sorted(manifest.metadata.items())),
+    }
+
+
+def _validate_reference_snapshot(
+    document: Mapping[str, Any],
+    namespace: str,
+    key: str,
+) -> None:
+    if (
+        document.get("schema_version") != REFERENCE_SCHEMA
+        or document.get("namespace") != namespace
+        or document.get("key") != key
+        or not isinstance(document.get("manifest_id"), str)
+        or _SHA256.fullmatch(str(document["manifest_id"])) is None
+        or not isinstance(document.get("updated_at_utc"), str)
+    ):
+        raise ArtifactIntegrityError("artifact reference snapshot is invalid")
+
+
+def _read_json_document(path: Path, *, unavailable_message: str) -> dict[str, Any]:
+    try:
+        metadata = path.lstat()
+        if (
+            stat.S_ISLNK(metadata.st_mode)
+            or not stat.S_ISREG(metadata.st_mode)
+            or not 0 < metadata.st_size <= _MAX_MANIFEST_BYTES
+        ):
+            raise ArtifactIntegrityError("artifact metadata path is invalid")
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise ArtifactStoreUnavailable(unavailable_message) from exc
+    except OSError as exc:
+        raise ArtifactStoreUnavailable(unavailable_message) from exc
+    except json.JSONDecodeError as exc:
+        raise ArtifactIntegrityError("artifact metadata is not valid JSON") from exc
+    if not isinstance(value, dict):
+        raise ArtifactIntegrityError("artifact metadata is not an object")
+    return value
+
+
+def _write_bytes_atomic(path: Path, payload: bytes, *, mode: int | None = None) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
+    try:
+        with temporary.open("xb") as stream:
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+        if mode is not None:
+            os.chmod(temporary, mode)
+        os.replace(temporary, path)
+    finally:
+        temporary.unlink(missing_ok=True)
+
+
+def _canonical_json(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    ).encode("utf-8")
+
+
+def _validate_component(value: str, label: str) -> None:
+    if not isinstance(value, str) or _SAFE_COMPONENT.fullmatch(value) is None:
+        raise ValueError(f"{label} is invalid")
+
+
+def _validate_sha256(value: str) -> None:
+    if not isinstance(value, str) or _SHA256.fullmatch(value) is None:
+        raise ValueError("artifact SHA-256 is invalid")

src/k1link/compute/integrated_perception.py

@@ -22,6 +22,11 @@ import numpy as np
 import rerun as rr
 from rerun.components import FillMode
 
+from k1link.artifact_gateway import (
+    ArtifactGateway,
+    ArtifactNotFound,
+    ArtifactStoreUnavailable,
+)
 from k1link.artifacts import write_json_atomic
 from k1link.sessions import SessionIntegrityError
 
@@ -377,12 +382,14 @@ class IntegratedPerceptionOverlayStore:
         lidar_packs_root: Path,
         cache_root: Path,
         ffmpeg_path: Path,
+        artifact_gateway: ArtifactGateway | None = None,
     ) -> None:
         self.jobs_root = jobs_root.expanduser().absolute()
         self.results_root = results_root.expanduser().absolute()
         self.lidar_packs_root = lidar_packs_root.expanduser().absolute()
         self.cache_root = cache_root.expanduser().absolute()
         self.ffmpeg_path = ffmpeg_path.expanduser().absolute()
+        self.artifact_gateway = artifact_gateway
         self._catalog_lock = threading.Lock()
         self._descriptor_catalog_generation: tuple[int, int] | None = None
         self._descriptors_by_session: dict[str, tuple[_ResultDescriptor, ...]] = {}
@@ -444,6 +451,15 @@ class IntegratedPerceptionOverlayStore:
         self._set_status(key, state="preparing", phase="cache-lookup")
         try:
             with self._single_flight(key):
+                central = self._read_gateway_cache(session_id, recording_id)
+                if central is not None:
+                    self._set_status(
+                        key,
+                        state="ready",
+                        phase="ready",
+                        byte_length=central.byte_length,
+                    )
+                    return central
                 cached = self._read_admitted_cache(session_id, recording_id)
                 if cached is not None:
                     self._set_status(
@@ -534,6 +550,50 @@ class IntegratedPerceptionOverlayStore:
             self._set_status(key, state="error", phase="error")
             raise
 
+    def _read_gateway_cache(
+        self,
+        session_id: str,
+        recording_id: str,
+    ) -> RecordedPerceptionOverlayArtifact | None:
+        if self.artifact_gateway is None:
+            return None
+        role = f"integrated-overlay:{recording_id}"
+        try:
+            resolved = self.artifact_gateway.resolve_role("sessions", session_id, role)
+        except (ArtifactNotFound, ArtifactStoreUnavailable):
+            return None
+        expected_result_id = resolved.manifest.metadata.get("integrated-result-id")
+        if (
+            resolved.member.media_type != "application/vnd.rerun.rrd"
+            or expected_result_id is None
+            or _SAFE_RESULT_ID.fullmatch(expected_result_id) is None
+        ):
+            raise RecordedPerceptionOverlayError(
+                "central integrated perception artifact metadata is invalid"
+            )
+        try:
+            metadata = resolved.path.lstat()
+            with resolved.path.open("rb") as stream:
+                magic = stream.read(4)
+        except OSError as exc:
+            raise RecordedPerceptionOverlayError(
+                "central integrated perception artifact became unavailable"
+            ) from exc
+        if (
+            stat.S_ISLNK(metadata.st_mode)
+            or not stat.S_ISREG(metadata.st_mode)
+            or metadata.st_size != resolved.member.byte_length
+            or magic != b"RRF2"
+        ):
+            raise RecordedPerceptionOverlayError(
+                "central integrated perception artifact is invalid"
+            )
+        return RecordedPerceptionOverlayArtifact(
+            path=resolved.path.resolve(strict=True),
+            byte_length=resolved.member.byte_length,
+            sha256=resolved.member.sha256,
+        )
+
     def status(self, session_id: str, *, recording_id: str) -> dict[str, Any]:
         if (
             _SAFE_RECORDING_ID.fullmatch(session_id) is None

src/k1link/device_plugins/xgrids_k1/cli.py

@@ -16,6 +16,7 @@ from rich.console import Console
 from rich.table import Table
 
 from k1link import __version__
+from k1link.artifact_cli import app as artifact_app
 from k1link.artifacts import write_json_atomic
 from k1link.compute import (
     prepare_camera_compute_job,
@@ -90,6 +91,7 @@ app.add_typer(analyze_app, name="analyze")
 app.add_typer(authority_app, name="authority")
 app.add_typer(compute_app, name="compute")
 app.add_typer(lab_app, name="lab")
+app.add_typer(artifact_app, name="artifact")
 
 
 class ToolStatus(TypedDict):

src/k1link/sessions/recording.py

@@ -16,6 +16,13 @@ from pathlib import Path
 from typing import Any, cast
 from uuid import uuid4
 
+from k1link.artifact_gateway import (
+    ArtifactGateway,
+    ArtifactGatewayError,
+    ArtifactNotFound,
+    ArtifactStoreUnavailable,
+)
+
 from .models import ReplayArtifact, ReplayCommand
 from .plugin_contract import (
     PluginRecordingExportCancelled,
@@ -139,6 +146,7 @@ class SessionRecordingMaterializer:
         exporters: Mapping[str, RecordingExporter] | None = None,
         cache_max_bytes: int | None = None,
         free_space_reserve_bytes: int | None = None,
+        artifact_gateway: ArtifactGateway | None = None,
     ) -> None:
         private_root = data_dir.expanduser().resolve()
         private_root.mkdir(mode=0o700, parents=True, exist_ok=True)
@@ -166,6 +174,7 @@ class SessionRecordingMaterializer:
             environment_name="MISSIONCORE_RRD_FREE_SPACE_RESERVE_BYTES",
             default=DEFAULT_FREE_SPACE_RESERVE_BYTES,
         )
+        self.artifact_gateway = artifact_gateway
         self._locks_guard = threading.Lock()
         self._session_locks: dict[str, threading.Lock] = {}
         self._memory_guard = threading.Lock()
@@ -349,6 +358,11 @@ class SessionRecordingMaterializer:
                 if cached is not None:
                     _report_progress(progress_callback, "ready", 1.0)
                     return cached
+                _report_progress(progress_callback, "restoring", 0.15)
+                restored = self._restore_gateway_recording_locked(session_id, source)
+                if restored is not None:
+                    _report_progress(progress_callback, "ready", 1.0)
+                    return restored
                 _report_progress(progress_callback, "exporting", 0.2)
                 return self._export_recording_locked(
                     session_id,
@@ -373,13 +387,114 @@ class SessionRecordingMaterializer:
                     self._scavenge_export_artifacts_locked()
                     source = _validate_source(command)
                     recording = self._load_cached_recording(session_id, source, pin=True)
+                    if recording is None:
+                        recording = self._restore_gateway_recording_locked(session_id, source)
                     if recording is None:
                         recording = self._export_recording_locked(session_id, source)
-                        with self._cache_guard:
+                    with self._cache_guard:
-                            self._increment_pin_locked(session_id)
+                        self._increment_pin_locked(session_id)
 
         return recording, self._release_callback(session_id)
 
+    def _restore_gateway_recording_locked(
+        self,
+        session_id: str,
+        source: _ValidatedSource,
+    ) -> MaterializedRecording | None:
+        if self.artifact_gateway is None:
+            return None
+        try:
+            resolved = self.artifact_gateway.resolve_role(
+                "sessions",
+                session_id,
+                "base-rrd",
+            )
+        except (ArtifactNotFound, ArtifactStoreUnavailable):
+            return None
+        except ArtifactGatewayError as exc:
+            raise RecordingMaterializationError(
+                "central recording artifact failed validation"
+            ) from exc
+        metadata = resolved.manifest.metadata
+        if (
+            resolved.manifest.artifact_type != "recorded-session"
+            or resolved.manifest.subject_id != session_id
+            or resolved.member.media_type != RERUN_RECORDING_MEDIA_TYPE
+            or metadata.get("base-timeline") != RERUN_SESSION_TIMELINE
+        ):
+            raise RecordingMaterializationError("central recording manifest is invalid")
+        source_sha256 = metadata.get("base-source-sha256")
+        try:
+            timeline_start_ns = int(metadata["base-timeline-start-ns"])
+            timeline_end_ns = int(metadata["base-timeline-end-ns"])
+        except (KeyError, TypeError, ValueError) as exc:
+            raise RecordingMaterializationError(
+                "central recording timeline metadata is invalid"
+            ) from exc
+        artifact_digests = _validated_artifact_digests(source)
+        if (
+            not isinstance(source_sha256, str)
+            or source_sha256 != artifact_digests[source.primary_artifact_id]
+            or timeline_start_ns < 0
+            or timeline_end_ns < timeline_start_ns
+        ):
+            raise RecordingMaterializationError(
+                "central recording does not match the local source evidence"
+            )
+        with self._cache_guard:
+            session_root, recording_path, sidecar_path = self._cache_paths(session_id)
+            self._ensure_cache_capacity(
+                required_bytes=resolved.member.byte_length + 4 * 1024,
+                protected_session_id=session_id,
+            )
+            candidate = session_root / f".scene.{uuid4().hex}.candidate.rrd"
+            try:
+                digest = hashlib.sha256()
+                byte_length = 0
+                with resolved.path.open("rb") as source_stream, candidate.open("xb") as output:
+                    while chunk := source_stream.read(1024 * 1024):
+                        output.write(chunk)
+                        digest.update(chunk)
+                        byte_length += len(chunk)
+                    output.flush()
+                    os.fsync(output.fileno())
+                if (
+                    digest.hexdigest() != resolved.member.sha256
+                    or byte_length != resolved.member.byte_length
+                ):
+                    raise RecordingMaterializationError(
+                        "central recording changed during local restoration"
+                    )
+                _chmod_best_effort(candidate, 0o600)
+                os.replace(candidate, recording_path)
+                _fsync_directory(session_root)
+                recording_stat = _regular_file_stat(recording_path, "derived recording")
+                recording = MaterializedRecording(
+                    session_id=session_id,
+                    path=recording_path,
+                    media_type=RERUN_RECORDING_MEDIA_TYPE,
+                    byte_length=byte_length,
+                    sha256=resolved.member.sha256,
+                    source_sha256=source_sha256,
+                    timeline=RERUN_SESSION_TIMELINE,
+                    timeline_start_ns=timeline_start_ns,
+                    timeline_end_ns=timeline_end_ns,
+                )
+                _write_json_atomic(
+                    sidecar_path,
+                    _cache_document(recording, source, recording_stat),
+                )
+                _chmod_best_effort(sidecar_path, 0o600)
+                self._remember(recording, source, recording_stat)
+                _touch_lru(session_root)
+                return recording
+            except OSError as exc:
+                raise RecordingMaterializationError(
+                    "central recording could not be restored locally"
+                ) from exc
+            finally:
+                candidate.unlink(missing_ok=True)
+
     def _release_callback(self, session_id: str) -> Callable[[], None]:
         released = False
         release_guard = threading.Lock()

src/k1link/web/app.py

@@ -15,6 +15,7 @@ from fastapi.staticfiles import StaticFiles
 from pydantic import ValidationError
 
 from k1link import __version__
+from k1link.artifact_gateway import configured_artifact_gateway
 from k1link.compute import (
     IntegratedPerceptionOverlayStore,
     RecordedCalibratedFusionStore,
@@ -88,9 +89,11 @@ plugin_environment = load_installed_device_plugins(REPOSITORY_ROOT)
 plugin_catalog: DevicePluginCatalog = plugin_environment.catalog
 plugin_dispatcher: DevicePluginDispatcher = plugin_environment.dispatcher
 session_store = SessionStore(REPOSITORY_ROOT)
+session_artifact_gateway = configured_artifact_gateway(session_store.data_dir)
 session_recording_materializer = SessionRecordingMaterializer(
     session_store.data_dir,
     exporters=plugin_environment.recording_exporters,
+    artifact_gateway=session_artifact_gateway,
 )
 session_recorded_media_inspector = RecordedMediaInspector(
     session_store.data_dir / "recorded-media-preparations"
@@ -136,6 +139,7 @@ session_integrated_perception_store = (
         ),
         cache_root=session_store.data_dir / "integrated-perception-overlays",
         ffmpeg_path=_ffmpeg,
+        artifact_gateway=session_artifact_gateway,
     )
     if _ffmpeg is not None
     else None

tests/test_artifact_gateway.py

@@ -0,0 +1,209 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from k1link.artifact_gateway import (
+    ArtifactCacheCapacityError,
+    ArtifactGateway,
+    ArtifactIntegrityError,
+    ArtifactStoreUnavailable,
+    CentralArtifactStore,
+    LocalArtifactCache,
+)
+
+
+def _gateway(
+    tmp_path: Path,
+    *,
+    max_bytes: int = 1024,
+) -> tuple[ArtifactGateway, Path, Path]:
+    central_root = tmp_path / "central"
+    local_root = tmp_path / "local"
+    return (
+        ArtifactGateway(
+            CentralArtifactStore(central_root, create=True),
+            LocalArtifactCache(
+                local_root,
+                max_bytes=max_bytes,
+                free_space_reserve_bytes=0,
+            ),
+        ),
+        central_root,
+        local_root,
+    )
+
+
+def test_gateway_publishes_deduplicated_objects_and_resolves_through_local_cache(
+    tmp_path: Path,
+) -> None:
+    gateway, central_root, _local_root = _gateway(tmp_path)
+    first = tmp_path / "first.rrd"
+    duplicate = tmp_path / "duplicate.rrd"
+    first.write_bytes(b"RRF2same-content")
+    duplicate.write_bytes(first.read_bytes())
+
+    manifest = gateway.publish(
+        namespace="sessions",
+        key="session-1",
+        artifact_type="recorded-session",
+        subject_id="session-1",
+        sources=(
+            ("base-rrd", "application/vnd.rerun.rrd", first),
+            ("overlay-rrd", "application/vnd.rerun.rrd", duplicate),
+        ),
+        metadata={"display-name": "Session 1"},
+    )
+
+    assert len(manifest.members) == 2
+    assert manifest.members[0].sha256 == manifest.members[1].sha256
+    objects = tuple((central_root / "objects" / "sha256").glob("*/*"))
+    assert len(objects) == 1
+
+    cold = gateway.resolve_role("sessions", "session-1", "overlay-rrd")
+    assert cold.path.read_bytes() == first.read_bytes()
+    assert cold.cache_hit is False
+    assert cold.central_available is True
+
+    warm = gateway.resolve_role("sessions", "session-1", "overlay-rrd")
+    assert warm.path == cold.path
+    assert warm.cache_hit is True
+
+
+def test_persistent_pin_survives_restart_and_resolves_with_central_offline(
+    tmp_path: Path,
+) -> None:
+    gateway, central_root, local_root = _gateway(tmp_path)
+    source = tmp_path / "overlay.rrd"
+    source.write_bytes(b"RRF2offline-ready")
+    gateway.publish(
+        namespace="sessions",
+        key="session-1",
+        artifact_type="recorded-session",
+        subject_id="session-1",
+        sources=(("overlay-rrd", "application/vnd.rerun.rrd", source),),
+    )
+    gateway.pin_reference("sessions", "session-1", pin_id="session-1")
+    offline_root = tmp_path / "central-offline"
+    os.replace(central_root, offline_root)
+
+    restarted = ArtifactGateway(
+        gateway.store,
+        LocalArtifactCache(
+            local_root,
+            max_bytes=1024,
+            free_space_reserve_bytes=0,
+        ),
+    )
+
+    resolved = restarted.resolve_role("sessions", "session-1", "overlay-rrd")
+    assert resolved.path.read_bytes() == source.read_bytes()
+    assert resolved.cache_hit is True
+    assert resolved.central_available is False
+    status = restarted.cache.status()
+    assert status.pinned_object_count == 1
+    assert status.pinned_bytes == len(source.read_bytes())
+
+
+def test_offline_cache_miss_is_explicit(tmp_path: Path) -> None:
+    gateway, central_root, local_root = _gateway(tmp_path)
+    source = tmp_path / "overlay.rrd"
+    source.write_bytes(b"RRF2not-fetched")
+    gateway.publish(
+        namespace="sessions",
+        key="session-1",
+        artifact_type="recorded-session",
+        subject_id="session-1",
+        sources=(("overlay-rrd", "application/vnd.rerun.rrd", source),),
+    )
+    os.replace(central_root, tmp_path / "central-offline")
+
+    restarted = ArtifactGateway(
+        gateway.store,
+        LocalArtifactCache(local_root, max_bytes=1024, free_space_reserve_bytes=0),
+    )
+
+    with pytest.raises(ArtifactStoreUnavailable, match="offline cache"):
+        restarted.resolve_role("sessions", "session-1", "overlay-rrd")
+
+
+def test_unpinned_lru_is_evicted_but_pinned_object_is_preserved(tmp_path: Path) -> None:
+    gateway, _central_root, _local_root = _gateway(tmp_path, max_bytes=18)
+    first = tmp_path / "first.bin"
+    second = tmp_path / "second.bin"
+    third = tmp_path / "third.bin"
+    first.write_bytes(b"a" * 9)
+    second.write_bytes(b"b" * 9)
+    third.write_bytes(b"c" * 9)
+    for key, source in (("first", first), ("second", second), ("third", third)):
+        gateway.publish(
+            namespace="objects",
+            key=key,
+            artifact_type="test-object",
+            subject_id=key,
+            sources=(("payload", "application/octet-stream", source),),
+        )
+
+    gateway.pin_reference("objects", "first", pin_id="keep-first")
+    second_resolved = gateway.resolve_role("objects", "second", "payload")
+    assert second_resolved.path.is_file()
+    gateway.resolve_role("objects", "third", "payload")
+
+    first_member = gateway.store.resolve_reference("objects", "first").member("payload")
+    second_member = gateway.store.resolve_reference("objects", "second").member("payload")
+    third_member = gateway.store.resolve_reference("objects", "third").member("payload")
+    assert gateway.cache.get(first_member) is not None
+    assert gateway.cache.get(second_member) is None
+    assert gateway.cache.get(third_member) is not None
+
+
+def test_cache_refuses_object_larger_than_its_quota(tmp_path: Path) -> None:
+    gateway, _central_root, _local_root = _gateway(tmp_path, max_bytes=8)
+    source = tmp_path / "large.bin"
+    source.write_bytes(b"x" * 9)
+    gateway.publish(
+        namespace="objects",
+        key="large",
+        artifact_type="test-object",
+        subject_id="large",
+        sources=(("payload", "application/octet-stream", source),),
+    )
+
+    with pytest.raises(ArtifactCacheCapacityError, match="exceeds"):
+        gateway.resolve_role("objects", "large", "payload")
+
+
+def test_same_size_local_corruption_is_detected_after_stat_identity_changes(
+    tmp_path: Path,
+) -> None:
+    gateway, _central_root, _local_root = _gateway(tmp_path)
+    source = tmp_path / "payload.bin"
+    source.write_bytes(b"original")
+    gateway.publish(
+        namespace="objects",
+        key="payload",
+        artifact_type="test-object",
+        subject_id="payload",
+        sources=(("payload", "application/octet-stream", source),),
+    )
+    resolved = gateway.resolve_role("objects", "payload", "payload")
+    resolved.path.write_bytes(b"tampered")
+    os.utime(resolved.path, ns=(resolved.path.stat().st_atime_ns, 1))
+
+    member = gateway.store.resolve_reference("objects", "payload").member("payload")
+    assert gateway.cache.get(member) is None
+    repaired = gateway.resolve_role("objects", "payload", "payload")
+    assert repaired.path.read_bytes() == b"original"
+
+
+def test_store_rejects_corrupt_existing_content_addressed_object(tmp_path: Path) -> None:
+    store = CentralArtifactStore(tmp_path / "central", create=True)
+    source = tmp_path / "source.bin"
+    source.write_bytes(b"original")
+    published = store.publish_file(source)
+    store.object_path(published.sha256).write_bytes(b"tampered")
+
+    with pytest.raises(ArtifactIntegrityError, match="corrupt"):
+        store.publish_file(source)

tests/test_e10_integrated_perception.py

@@ -9,11 +9,18 @@ import time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from types import SimpleNamespace
+from typing import cast
 
 import numpy as np
 import pytest
 
 import k1link.compute.integrated_perception as integrated_module
+from k1link.artifact_gateway import (
+    ArtifactGateway,
+    ArtifactManifest,
+    ArtifactMember,
+    ResolvedArtifact,
+)
 from k1link.compute.integrated_perception import (
     IntegratedPerceptionOverlayStore,
     _CuboidPresentationState,
@@ -227,6 +234,75 @@ def test_integrated_overlay_serializes_heavy_materialization_across_sessions(
     assert maximum_active == 1
 
 
+def test_integrated_overlay_uses_verified_central_cache_before_local_result_scan(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    for name in ("jobs", "results", "lidar-packs"):
+        (tmp_path / name).mkdir()
+    recording_id = "recording-1"
+    result_id = f"e10-integrated-perception-{'a' * 64}"
+    payload = b"RRF2central-overlay"
+    artifact_path = tmp_path / "artifact-cache" / "overlay.rrd"
+    artifact_path.parent.mkdir()
+    artifact_path.write_bytes(payload)
+    member = ArtifactMember(
+        role=f"integrated-overlay:{recording_id}",
+        media_type="application/vnd.rerun.rrd",
+        sha256=hashlib.sha256(payload).hexdigest(),
+        byte_length=len(payload),
+    )
+    manifest = ArtifactManifest(
+        manifest_id="b" * 64,
+        artifact_type="recorded-session",
+        subject_id="session-1",
+        created_at_utc="2026-07-30T00:00:00Z",
+        members=(member,),
+        metadata={"integrated-result-id": result_id},
+    )
+
+    class FakeGateway:
+        def resolve_role(self, namespace: str, key: str, role: str) -> ResolvedArtifact:
+            assert (namespace, key, role) == (
+                "sessions",
+                "session-1",
+                f"integrated-overlay:{recording_id}",
+            )
+            return ResolvedArtifact(
+                manifest=manifest,
+                member=member,
+                path=artifact_path,
+                cache_hit=False,
+                central_available=True,
+            )
+
+    store = IntegratedPerceptionOverlayStore(
+        jobs_root=tmp_path / "jobs",
+        results_root=tmp_path / "results",
+        lidar_packs_root=tmp_path / "lidar-packs",
+        cache_root=tmp_path / "cache",
+        ffmpeg_path=tmp_path / "ffmpeg",
+        artifact_gateway=cast(ArtifactGateway, FakeGateway()),
+    )
+    monkeypatch.setattr(
+        store,
+        "_latest_descriptor",
+        lambda _session_id: (_ for _ in ()).throw(
+            AssertionError("central cache hit must not scan local results")
+        ),
+    )
+
+    artifact = store.materialize(
+        "session-1",
+        application_id="nodedc_mission_core_recorded",
+        recording_id=recording_id,
+    )
+
+    assert artifact is not None
+    assert artifact.path == artifact_path
+    assert artifact.sha256 == member.sha256
+
+
 def test_e10_profile_pins_integrated_realtime_budget() -> None:
     _fusion, runner = _worker_modules()
     profile_path = (

tests/test_session_recording.py

@@ -13,6 +13,11 @@ from types import SimpleNamespace
 import pytest
 
 import k1link.sessions.recording as recording_module
+from k1link.artifact_gateway import (
+    ArtifactGateway,
+    CentralArtifactStore,
+    LocalArtifactCache,
+)
 from k1link.sessions.models import ReplayArtifact, ReplayCommand
 from k1link.sessions.recording import (
     CACHE_SCHEMA,
@@ -131,6 +136,54 @@ def test_materializer_reuses_only_a_digest_validated_private_cache(tmp_path: Pat
     assert exporter.calls == 1
 
 
+def test_materializer_restores_exact_central_recording_without_export(
+    tmp_path: Path,
+) -> None:
+    command = _command(tmp_path / "source")
+    central_recording = tmp_path / "central.rrd"
+    central_recording.write_bytes(b"RRF2centrally-prepared-recording")
+    gateway = ArtifactGateway(
+        CentralArtifactStore(tmp_path / "central-store", create=True),
+        LocalArtifactCache(
+            tmp_path / "artifact-cache",
+            max_bytes=1024 * 1024,
+            free_space_reserve_bytes=0,
+        ),
+    )
+    gateway.publish(
+        namespace="sessions",
+        key=command.session_id,
+        artifact_type="recorded-session",
+        subject_id=command.session_id,
+        sources=(
+            ("base-rrd", RERUN_RECORDING_MEDIA_TYPE, central_recording),
+        ),
+        metadata={
+            "base-source-sha256": _sha256(command.primary_artifact.path),
+            "base-timeline": "session_time",
+            "base-timeline-start-ns": "0",
+            "base-timeline-end-ns": "2500000000",
+        },
+    )
+    exporter = FakeExporter()
+    materializer = SessionRecordingMaterializer(
+        tmp_path / "private",
+        exporter=exporter,
+        artifact_gateway=gateway,
+    )
+
+    restored = materializer.materialize(command)
+
+    assert exporter.calls == 0
+    assert restored.path.is_relative_to(materializer.recordings_root)
+    assert restored.path.read_bytes() == central_recording.read_bytes()
+    assert restored.sha256 == _sha256(central_recording)
+    assert restored.source_sha256 == _sha256(command.primary_artifact.path)
+    assert restored.timeline_end_ns == 2_500_000_000
+    assert materializer.materialize(command) == restored
+    assert exporter.calls == 0
+
+
 def test_default_recording_cache_has_no_application_byte_quota(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,