runtime: canonicalize snapshot store semantics and revalidate persisted snapshots

src/open_range/builder/snapshot_store.py

@@ -46,7 +46,6 @@ class SnapshotStore:
             The snapshot ID string.
         """
         if snapshot_id is None:
-            hosts = snapshot.topology.get("hosts", [])
             vuln_types = [v.type for v in snapshot.truth_graph.vulns]
             snapshot_id = (
                 f"snap_{'_'.join(vuln_types[:3])}"
@@ -63,21 +62,7 @@ class SnapshotStore:
         )
 
         # Write metadata sidecar for fast listing
-        meta = {
-            "snapshot_id": snapshot_id,
-            "vuln_classes": [v.type for v in snapshot.truth_graph.vulns],
-            "golden_path_steps": len(snapshot.golden_path),
-            "flag_count": len(snapshot.flags),
-            "npc_count": len(snapshot.npc_personas),
-            "has_compose": bool(snapshot.compose),
-            "has_payload_files": bool(snapshot.files),
-            "live_validated": bool(snapshot.topology.get("live_validated", False)),
-            "parent_snapshot_id": snapshot.lineage.parent_snapshot_id,
-            "root_snapshot_id": snapshot.lineage.root_snapshot_id,
-            "generation_depth": snapshot.lineage.generation_depth,
-            "mutation_summary": list(snapshot.lineage.mutation_summary),
-            "stored_at": time.time(),
-        }
+        meta = self._metadata_from_snapshot(snapshot_id, snapshot)
         meta_path = snap_dir / "metadata.json"
         meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8")
 
@@ -113,38 +98,69 @@ class SnapshotStore:
         else:  # latest -- sort by parent dir mtime
             chosen = max(spec_files, key=lambda p: p.stat().st_mtime)
 
-        raw = json.loads(chosen.read_text(encoding="utf-8"))
         return StoredSnapshot(
             snapshot_id=chosen.parent.name,
-            snapshot=SnapshotSpec.model_validate(raw),
+            snapshot=self._load_spec(chosen),
         )
 
     async def list_entries(self) -> list[StoredSnapshot]:
         """Return every stored snapshot plus its persisted ID."""
         entries: list[StoredSnapshot] = []
         for spec_path in sorted(self.store_dir.glob("*/spec.json")):
-            raw = json.loads(spec_path.read_text(encoding="utf-8"))
             entries.append(
                 StoredSnapshot(
                     snapshot_id=spec_path.parent.name,
-                    snapshot=SnapshotSpec.model_validate(raw),
+                    snapshot=self._load_spec(spec_path),
                 )
             )
         return entries
 
+    async def count_entries(self) -> int:
+        """Return canonical snapshot count based on persisted specs."""
+        return len(await self.list_entries())
+
     async def list_snapshots(self) -> list[dict[str, Any]]:
         """List all snapshots with their metadata.
 
         Returns:
             List of metadata dicts, sorted by stored_at descending.
         """
+        entries = await self.list_entries()
+        spec_ids = {entry.snapshot_id for entry in entries}
         results: list[dict[str, Any]] = []
-        for meta_path in self.store_dir.glob("*/metadata.json"):
+        for entry in entries:
+            meta_path = self.store_dir / entry.snapshot_id / "metadata.json"
+            existing_meta: dict[str, Any] | None = None
             try:
-                meta = json.loads(meta_path.read_text(encoding="utf-8"))
-                results.append(meta)
+                if meta_path.exists():
+                    loaded = json.loads(meta_path.read_text(encoding="utf-8"))
+                    if isinstance(loaded, dict):
+                        existing_meta = loaded
+                    else:
+                        logger.warning(
+                            "Repairing metadata sidecar with non-object payload: %s",
+                            meta_path,
+                        )
             except (json.JSONDecodeError, OSError) as exc:
-                logger.warning("Skipping corrupt metadata: %s (%s)", meta_path, exc)
+                logger.warning("Repairing corrupt metadata: %s (%s)", meta_path, exc)
+
+            stored_at = existing_meta.get("stored_at") if existing_meta else None
+            canonical = self._metadata_from_snapshot(
+                entry.snapshot_id,
+                entry.snapshot,
+                stored_at=stored_at if isinstance(stored_at, (int, float)) else None,
+            )
+            results.append(canonical)
+
+            if existing_meta != canonical:
+                try:
+                    meta_path.write_text(json.dumps(canonical, indent=2), encoding="utf-8")
+                except OSError as exc:
+                    logger.warning("Failed to repair metadata sidecar %s (%s)", meta_path, exc)
+
+        for meta_path in self.store_dir.glob("*/metadata.json"):
+            if meta_path.parent.name not in spec_ids:
+                logger.warning("Ignoring orphan metadata without spec.json: %s", meta_path)
 
         results.sort(key=lambda m: m.get("stored_at", 0), reverse=True)
         return results
@@ -158,8 +174,7 @@ class SnapshotStore:
         spec_path = self.store_dir / snapshot_id / "spec.json"
         if not spec_path.exists():
             raise FileNotFoundError(f"Snapshot not found: {snapshot_id}")
-        raw = json.loads(spec_path.read_text(encoding="utf-8"))
-        return SnapshotSpec.model_validate(raw)
+        return self._load_spec(spec_path)
 
     async def get_entry(self, snapshot_id: str) -> StoredSnapshot:
         """Load a specific snapshot plus its ID."""
@@ -167,3 +182,34 @@ class SnapshotStore:
             snapshot_id=snapshot_id,
             snapshot=await self.get(snapshot_id),
         )
+
+    @staticmethod
+    def _metadata_from_snapshot(
+        snapshot_id: str,
+        snapshot: SnapshotSpec,
+        *,
+        stored_at: float | None = None,
+    ) -> dict[str, Any]:
+        return {
+            "snapshot_id": snapshot_id,
+            "vuln_classes": [v.type for v in snapshot.truth_graph.vulns],
+            "golden_path_steps": len(snapshot.golden_path),
+            "flag_count": len(snapshot.flags),
+            "npc_count": len(snapshot.npc_personas),
+            "has_compose": bool(snapshot.compose),
+            "has_payload_files": bool(snapshot.files),
+            "live_validated": bool(snapshot.topology.get("live_validated", False)),
+            "parent_snapshot_id": snapshot.lineage.parent_snapshot_id,
+            "root_snapshot_id": snapshot.lineage.root_snapshot_id,
+            "generation_depth": snapshot.lineage.generation_depth,
+            "mutation_summary": list(snapshot.lineage.mutation_summary),
+            "stored_at": float(time.time() if stored_at is None else stored_at),
+        }
+
+    @staticmethod
+    def _load_spec(spec_path: Path) -> SnapshotSpec:
+        try:
+            raw = json.loads(spec_path.read_text(encoding="utf-8"))
+            return SnapshotSpec.model_validate(raw)
+        except Exception as exc:  # noqa: BLE001
+            raise ValueError(f"invalid snapshot spec at {spec_path}: {exc}") from exc

src/open_range/server/runtime.py

@@ -66,6 +66,13 @@ _VALIDATOR_PROFILE_ALIASES = {
     "strict": "training",
 }
 _LIVE_VALIDATOR_PROFILES = {"training"}
+_PERSISTED_SNAPSHOT_VALIDATION_ALIASES = {
+    "none": "trust",
+    "disabled": "trust",
+    "off": "trust",
+    "revalidate": "offline",
+    "strict": "offline",
+}
 
 
 def _env_flag(name: str, default: bool = False) -> bool:
@@ -318,6 +325,17 @@ def _normalize_validator_profile(profile: str | None) -> str:
     return normalized
 
 
+def _normalize_persisted_snapshot_validation(policy: str | None) -> str:
+    normalized = (policy or "offline").strip().lower()
+    normalized = _PERSISTED_SNAPSHOT_VALIDATION_ALIASES.get(normalized, normalized)
+    if normalized not in {"trust", "offline"}:
+        raise ValueError(
+            f"Unsupported persisted snapshot validation policy {policy!r}. "
+            "Expected 'trust' or 'offline'."
+        )
+    return normalized
+
+
 def _graph_checks(manifest: dict[str, Any]) -> list[Any]:
     return [
         ManifestComplianceCheck(manifest),
@@ -392,6 +410,7 @@ class ManagedSnapshotRuntime:
         compose_runner: ComposeProjectRunner | None = None,
         live_validator: ValidatorGate | None = None,
         enable_patch_validation: bool = False,
+        persisted_snapshot_validation: str | None = None,
         mutation_policy: PopulationMutationPolicy | None = None,
     ) -> None:
         self.manifest_path = (
@@ -408,7 +427,16 @@ class ManagedSnapshotRuntime:
         self.validator_profile = _normalize_validator_profile(
             validator_profile or os.getenv("OPENRANGE_RUNTIME_VALIDATOR_PROFILE", "offline")
         )
+        self.persisted_snapshot_validation = _normalize_persisted_snapshot_validation(
+            persisted_snapshot_validation
+            or os.getenv("OPENRANGE_PERSISTED_SNAPSHOT_VALIDATION", "offline")
+        )
         self.validator = validator or _build_validator(self.validator_profile, self.manifest)
+        self.persisted_validator = (
+            _build_validator("offline", self.manifest)
+            if self.persisted_snapshot_validation == "offline"
+            else None
+        )
         self.renderer = SnapshotRenderer()
         self.curriculum = CurriculumTracker()
         self.pool_size = max(1, pool_size)
@@ -458,6 +486,10 @@ class ManagedSnapshotRuntime:
                 "OPENRANGE_ENABLE_PATCH_VALIDATION",
                 default=False,
             ),
+            persisted_snapshot_validation=os.getenv(
+                "OPENRANGE_PERSISTED_SNAPSHOT_VALIDATION",
+                "offline",
+            ),
         )
 
     @staticmethod
@@ -477,6 +509,7 @@ class ManagedSnapshotRuntime:
             if existing < self.pool_size:
                 self._top_up_pool(self.pool_size - existing)
             self._ensure_existing_artifacts()
+            self._revalidate_persisted_snapshots()
 
             available = self.snapshot_count()
             if available == 0:
@@ -528,6 +561,7 @@ class ManagedSnapshotRuntime:
             if alternative is not None:
                 stored = alternative
 
+        self._assert_persisted_snapshot_valid(stored.snapshot_id, stored.snapshot)
         result = RuntimeSnapshot(snapshot_id=stored.snapshot_id, snapshot=stored.snapshot)
         self._track_acquisition(result.snapshot_id)
         return result
@@ -544,14 +578,15 @@ class ManagedSnapshotRuntime:
         if not recent_ids:
             return set()
 
-        all_meta = self.list_snapshots()
-        meta_by_id = {m.get("snapshot_id"): m for m in all_meta}
+        entries = _run_coro_sync(self.store.list_entries())
+        by_id = {entry.snapshot_id: entry for entry in entries}
         vuln_types: set[str] = set()
         for sid in recent_ids:
-            meta = meta_by_id.get(sid)
-            if meta:
-                vuln_types.update(meta.get("vuln_classes", []))
+            entry = by_id.get(sid)
+            if entry:
+                vuln_types.update(v.type for v in entry.snapshot.truth_graph.vulns)
         return vuln_types
+
     def _is_diverse(self, snapshot: SnapshotSpec) -> bool:
         """Return True if *snapshot* has at least one vuln type not in recent history."""
         recent = self._recent_vuln_types()
@@ -569,32 +604,29 @@ class ManagedSnapshotRuntime:
         """Try to find a snapshot in the store whose vulns don't fully overlap."""
         from open_range.builder.snapshot_store import StoredSnapshot
 
-        all_meta = self.list_snapshots()
+        entries = _run_coro_sync(self.store.list_entries())
         recent = self._recent_vuln_types()
 
-        for meta in all_meta:
-            sid = meta.get("snapshot_id", "")
+        for entry in entries:
+            sid = entry.snapshot_id
             if sid == exclude_id:
                 continue
-            candidate_vulns = set(meta.get("vuln_classes", []))
+            candidate_vulns = {v.type for v in entry.snapshot.truth_graph.vulns}
             if not candidate_vulns or not candidate_vulns.issubset(recent):
-                try:
-                    entry = _run_coro_sync(self.store.get_entry(sid))
-                    return entry
-                except Exception:  # noqa: BLE001
-                    continue
+                return entry
         return None
 
     def get_snapshot(self, snapshot_id: str) -> RuntimeSnapshot:
         self.start()
         stored = _run_coro_sync(self.store.get_entry(snapshot_id))
+        self._assert_persisted_snapshot_valid(stored.snapshot_id, stored.snapshot)
         return RuntimeSnapshot(snapshot_id=stored.snapshot_id, snapshot=stored.snapshot)
 
     def list_snapshots(self) -> list[dict[str, Any]]:
         return _run_coro_sync(self.store.list_snapshots())
 
     def snapshot_count(self) -> int:
-        return len(self.list_snapshots())
+        return int(_run_coro_sync(self.store.count_entries()))
 
     def status(self) -> dict[str, Any]:
         return {
@@ -604,6 +636,7 @@ class ManagedSnapshotRuntime:
             "selection_strategy": self.selection_strategy,
             "parent_selection_strategy": self.parent_selection_strategy,
             "validator_profile": self.validator_profile,
+            "persisted_snapshot_validation": self.persisted_snapshot_validation,
             "refill_enabled": self.refill_enabled,
             "live_admission_enabled": self.live_admission_enabled,
             "snapshot_count": self.snapshot_count(),
@@ -666,30 +699,33 @@ class ManagedSnapshotRuntime:
             self._generate_and_store_snapshot()
 
     def _ensure_existing_artifacts(self) -> None:
-        for meta in self.list_snapshots():
-            snapshot_id = str(meta.get("snapshot_id", ""))
-            if not snapshot_id:
-                continue
+        for stored in _run_coro_sync(self.store.list_entries()):
+            snapshot_id = stored.snapshot_id
             artifacts_dir = self._artifacts_dir(snapshot_id)
             if artifacts_dir.exists():
                 continue
-            stored = _run_coro_sync(self.store.get_entry(snapshot_id))
             materialized = self._materialize_snapshot(stored.snapshot, snapshot_id)
             _run_coro_sync(self.store.store(materialized, snapshot_id=snapshot_id))
 
+    def _revalidate_persisted_snapshots(self) -> None:
+        if self.persisted_snapshot_validation == "trust":
+            return
+        for entry in _run_coro_sync(self.store.list_entries()):
+            self._assert_persisted_snapshot_valid(entry.snapshot_id, entry.snapshot)
+
+    def _assert_persisted_snapshot_valid(self, snapshot_id: str, snapshot: SnapshotSpec) -> None:
+        if self.persisted_validator is None:
+            return
+        result = _run_coro_sync(self.persisted_validator.validate(snapshot, ContainerSet()))
+        if result.passed:
+            return
+        raise RuntimeError(
+            "persisted snapshot failed startup revalidation "
+            f"({snapshot_id}): {self._validation_error(result)}"
+        )
+
     def _generate_and_store_snapshot(self) -> str:
         last_error: str | None = None
-        parent_snapshot: SnapshotSpec | None = None
-        parent_snapshot_id: str | None = None
-        existing = self.list_snapshots()
-        if existing:
-            parent_snapshot_id = str(existing[0].get("snapshot_id", "") or "")
-            if parent_snapshot_id:
-                try:
-                    parent_snapshot = _run_coro_sync(self.store.get(parent_snapshot_id))
-                except FileNotFoundError:
-                    parent_snapshot = None
-                    parent_snapshot_id = None
 
         for attempt in range(1, self.generation_retries + 1):
             context = self._build_context()

tests/test_builder.py

@@ -2,6 +2,7 @@
 
 import json
 import tempfile
+from pathlib import Path
 
 import pytest
 
@@ -542,6 +543,52 @@ async def test_snapshot_store_list():
         assert "snap_b" in ids
 
 
+@pytest.mark.asyncio
+async def test_snapshot_store_repairs_missing_metadata_from_spec():
+    from open_range.builder.snapshot_store import SnapshotStore
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        store = SnapshotStore(store_dir=tmpdir)
+        spec = SnapshotSpec(topology={"hosts": ["web"]})
+        await store.store(spec, snapshot_id="snap_a")
+
+        metadata_path = Path(tmpdir) / "snap_a" / "metadata.json"
+        metadata_path.unlink()
+
+        listing = await store.list_snapshots()
+        assert len(listing) == 1
+        assert listing[0]["snapshot_id"] == "snap_a"
+        assert metadata_path.exists()
+
+        selected = await store.select_entry(strategy="latest")
+        assert selected.snapshot_id == "snap_a"
+
+
+@pytest.mark.asyncio
+async def test_snapshot_store_ignores_orphan_metadata_without_spec():
+    from open_range.builder.snapshot_store import SnapshotStore
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        store = SnapshotStore(store_dir=tmpdir)
+        spec = SnapshotSpec(topology={"hosts": ["web"]})
+        await store.store(spec, snapshot_id="snap_real")
+
+        orphan_dir = Path(tmpdir) / "orphan_meta"
+        orphan_dir.mkdir(parents=True, exist_ok=True)
+        (orphan_dir / "metadata.json").write_text(
+            json.dumps({"snapshot_id": "orphan_meta", "stored_at": 9999999999}),
+            encoding="utf-8",
+        )
+
+        listing = await store.list_snapshots()
+        ids = {meta["snapshot_id"] for meta in listing}
+        assert ids == {"snap_real"}
+        assert await store.count_entries() == 1
+
+        selected = await store.select_entry(strategy="latest")
+        assert selected.snapshot_id == "snap_real"
+
+
 @pytest.mark.asyncio
 async def test_snapshot_store_get_by_id():
     from open_range.builder.snapshot_store import SnapshotStore

tests/test_runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 import pytest
@@ -70,6 +71,44 @@ class TestManagedSnapshotRuntime:
         finally:
             runtime.stop()
 
+    def test_start_revalidates_persisted_snapshots_by_default(self, tier1_manifest, tmp_path):
+        store_dir = tmp_path / "snapshots"
+
+        runtime = ManagedSnapshotRuntime(
+            manifest=tier1_manifest,
+            store_dir=store_dir,
+            pool_size=1,
+            refill_enabled=False,
+        )
+        runtime.start()
+        runtime.stop()
+
+        spec_path = next(store_dir.glob("*/spec.json"))
+        raw = json.loads(spec_path.read_text(encoding="utf-8"))
+        raw["truth_graph"]["vulns"] = []
+        raw["golden_path"] = []
+        raw["flags"] = []
+        spec_path.write_text(json.dumps(raw, indent=2), encoding="utf-8")
+
+        runtime = ManagedSnapshotRuntime(
+            manifest=tier1_manifest,
+            store_dir=store_dir,
+            pool_size=1,
+            refill_enabled=False,
+        )
+        with pytest.raises(RuntimeError, match="persisted snapshot failed startup revalidation"):
+            runtime.start()
+
+        trust_runtime = ManagedSnapshotRuntime(
+            manifest=tier1_manifest,
+            store_dir=store_dir,
+            pool_size=1,
+            refill_enabled=False,
+            persisted_snapshot_validation="trust",
+        )
+        trust_runtime.start()
+        trust_runtime.stop()
+
     def test_start_materializes_rendered_artifacts(self, tier1_manifest, tmp_path):
         runtime = ManagedSnapshotRuntime(
             manifest=tier1_manifest,