Merge remote-tracking branch 'origin/main' into codex/issue-77-20260308

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

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import inspect
 import logging
 import os
 
@@ -10,6 +11,24 @@ from fastapi import FastAPI
 logger = logging.getLogger(__name__)
 
 
+def _extract_openenv_server(fastapp: FastAPI) -> object | None:
+    """Best-effort extraction of OpenEnv's HTTPEnvServer from route closure."""
+    for route in fastapp.router.routes:
+        if getattr(route, "path", None) != "/ws":
+            continue
+        endpoint = getattr(route, "endpoint", None)
+        if endpoint is None:
+            continue
+        try:
+            closure = inspect.getclosurevars(endpoint)
+        except Exception:
+            continue
+        server = closure.nonlocals.get("self")
+        if server is not None and hasattr(server, "active_sessions"):
+            return server
+    return None
+
+
 def create_app() -> FastAPI:
     """Create the OpenRange app through the canonical OpenEnv factory."""
     from openenv.core.env_server import create_app as create_openenv_app
@@ -37,6 +56,9 @@ def create_app() -> FastAPI:
         RangeObservation,
         env_name="open_range",
     )
+    openenv_server = _extract_openenv_server(fastapp)
+    if openenv_server is not None:
+        fastapp.state.openenv_server = openenv_server
 
     # Mount custom Gradio dashboard at /web if gradio is available
     try:

src/open_range/server/console.py

@@ -47,10 +47,20 @@ def get_history(limit: int = 20) -> list[dict[str, Any]]:
 @console_router.get("/api/snapshot")
 async def api_snapshot(request: Request) -> JSONResponse:
     """Return current snapshot metadata (no truth graph or flags)."""
-    env = _get_env(request)
+    ctx = _get_env_context(request)
+    env = ctx["env"]
     snapshot = env.snapshot
     if snapshot is None:
-        return JSONResponse({"id": None, "tier": None, "hosts": [], "zones": {}, "vuln_count": 0})
+        return JSONResponse({
+            "id": None,
+            "tier": None,
+            "hosts": [],
+            "zones": {},
+            "vuln_count": 0,
+            "state_scope": ctx["state_scope"],
+            "session_id": ctx["session_id"],
+            "warning": ctx["warning"],
+        })
 
     topo = snapshot.topology if isinstance(snapshot.topology, dict) else {}
     hosts = topo.get("hosts", [])
@@ -64,19 +74,26 @@ async def api_snapshot(request: Request) -> JSONResponse:
         "hosts": hosts,
         "zones": zones,
         "vuln_count": vuln_count,
+        "state_scope": ctx["state_scope"],
+        "session_id": ctx["session_id"],
+        "warning": ctx["warning"],
     })
 
 
 @console_router.get("/api/episode")
 async def api_episode(request: Request) -> JSONResponse:
     """Return current episode state."""
-    env = _get_env(request)
+    ctx = _get_env_context(request)
+    env = ctx["env"]
     state = env.state
     return JSONResponse({
         "step_count": state.step_count,
         "flags_found": len(state.flags_found),
         "mode": state.mode,
         "services_status": state.services_status,
+        "state_scope": ctx["state_scope"],
+        "session_id": ctx["session_id"],
+        "warning": ctx["warning"],
     })
 
 
@@ -98,20 +115,70 @@ async def console_page() -> HTMLResponse:
 # ---------------------------------------------------------------------------
 
 
-def _get_env(request: Request) -> Any:
-    """Retrieve the RangeEnvironment from the app's state.
+def _get_env_context(request: Request) -> dict[str, Any]:
+    """Resolve the environment context used by the console endpoints.
 
-    The app.py startup stores the environment instance as ``app.state.env``.
-    If that attribute is missing we fall back to importing a fresh one.
+    Priority:
+    1. Active OpenEnv WebSocket session environment (session-scoped truth)
+    2. ``app.state.env`` fallback environment (global app scope)
+    3. Lazily created fallback environment (tests/dev)
     """
     app = request.app
+
+    server = getattr(app.state, "openenv_server", None)
+    sessions = getattr(server, "_sessions", None)
+    if isinstance(sessions, dict) and sessions:
+        if len(sessions) == 1:
+            session_id, env = next(iter(sessions.items()))
+            return {
+                "env": env,
+                "state_scope": "websocket_session",
+                "session_id": session_id,
+                "warning": None,
+            }
+
+        session_info = getattr(server, "_session_info", {})
+        selected_id = max(
+            sessions.keys(),
+            key=lambda sid: float(getattr(session_info.get(sid), "last_activity_at", 0.0) or 0.0),
+        )
+        return {
+            "env": sessions[selected_id],
+            "state_scope": "websocket_session",
+            "session_id": selected_id,
+            "warning": (
+                f"{len(sessions)} active sessions detected; "
+                f"showing the most recently active session ({selected_id})."
+            ),
+        }
+
     if hasattr(app.state, "env"):
-        return app.state.env
-    # Fallback: create an ephemeral environment (tests, etc.)
+        return {
+            "env": app.state.env,
+            "state_scope": "app_state_env",
+            "session_id": None,
+            "warning": (
+                "No active WebSocket session found; console is showing shared "
+                "app-state environment data."
+            ),
+        }
+
+    # Fallback: create an ephemeral environment (tests/dev)
     from open_range.server.environment import RangeEnvironment
+
     if not hasattr(app.state, "_fallback_env"):
         app.state._fallback_env = RangeEnvironment(docker_available=False)
-    return app.state._fallback_env
+    return {
+        "env": app.state._fallback_env,
+        "state_scope": "fallback_env",
+        "session_id": None,
+        "warning": "Console is using a fallback environment (no server session available).",
+    }
+
+
+def _get_env(request: Request) -> Any:
+    """Compatibility helper for callers that only need the env object."""
+    return _get_env_context(request)["env"]
 
 
 # ---------------------------------------------------------------------------

src/open_range/server/environment.py

@@ -1276,6 +1276,62 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         )
         return self._container_name(name)
 
+    def _topology_host_names(self) -> list[str]:
+        """Return deduplicated host names from the active snapshot topology."""
+        if not self._snapshot or not isinstance(self._snapshot.topology, dict):
+            return []
+        hosts = self._snapshot.topology.get("hosts", [])
+        names: list[str] = []
+        for host in hosts:
+            if isinstance(host, str):
+                candidate = host
+            elif isinstance(host, dict):
+                candidate = host.get("name") or host.get("hostname") or ""
+            else:
+                candidate = ""
+            name = str(candidate).strip()
+            if name and name not in names:
+                names.append(name)
+        return names
+
+    def _refresh_services_status(self) -> None:
+        """Refresh ``state.services_status`` from runtime/container health.
+
+        Availability reward should never rely on an empty status map after reset.
+        When health cannot be verified, host status is marked ``"unknown"``.
+        """
+        host_names = self._topology_host_names()
+        if not host_names:
+            self._state.services_status = {}
+            return
+
+        status_map = {host: "unknown" for host in host_names}
+
+        if self._execution_mode == "docker" and self._docker_available is not False:
+            client = self._get_docker()
+            if client is not None:
+                for host in host_names:
+                    container_name = self._container_name(host)
+                    try:
+                        container = client.containers.get(container_name)
+                        status_map[host] = str(getattr(container, "status", "unknown") or "unknown")
+                    except Exception:
+                        status_map[host] = "down"
+                self._state.services_status = status_map
+                return
+
+        if self._execution_mode == "subprocess" and self._snapshot and self._snapshot.services:
+            checks_by_host: dict[str, list[bool]] = {}
+            for svc in self._snapshot.services:
+                host = str(getattr(svc, "host", "") or "").strip()
+                if not host:
+                    continue
+                checks_by_host.setdefault(host, []).append(self._probe_readiness(svc.readiness))
+            for host, checks in checks_by_host.items():
+                status_map[host] = "healthy" if checks and all(checks) else "degraded"
+
+        self._state.services_status = status_map
+
     # -----------------------------------------------------------------
     # Core API
     # -----------------------------------------------------------------
@@ -1354,6 +1410,9 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         # Start NPC traffic for this episode
         self._start_npcs(self._snapshot)
 
+        # Prime service health map for availability reward grounding.
+        self._refresh_services_status()
+
         # Build initial briefing
         task = self._snapshot.task
         if isinstance(task, dict):
@@ -1439,6 +1498,7 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
 
         if cmd_name in meta_handlers:
             obs = meta_handlers[cmd_name](action)
+            self._refresh_services_status()
             obs = self._apply_rewards(action, obs)
             self._check_termination(obs)
             self._report_if_done(obs)
@@ -1484,6 +1544,7 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
 
         # Refresh NPC traffic log for reward computation
         self._refresh_npc_traffic_log()
+        self._refresh_services_status()
 
         # Build observation
         obs = RangeObservation(
@@ -1620,8 +1681,8 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
 
         In production (docker or subprocess mode with real infrastructure),
         queries the SIEM container for actual log-based alerts. Falls back
-        to synthetic alerts derived from ALL Red actions when SIEM queries
-        return nothing or in unit-test mock mode.
+        to synthetic alerts derived from Red action history when SIEM queries
+        return nothing or in unit-test mock mode (capped to recent 20 lines).
         """
         # Try real SIEM query in non-mock modes
         if self._docker_available is not False or self._execution_mode == "subprocess":
@@ -1629,7 +1690,23 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             if siem_alerts:
                 return siem_alerts
 
-        return []
+        # Fallback: synthesize alerts from recent Red actions so Blue still
+        # receives actionable signal in mock/degraded SIEM paths.
+        synthetic: list[str] = []
+        for record in self._red_history:
+            if record.get("type") in ("hallucinated_flag", "evidence"):
+                continue
+            command = str(record.get("command", "")).strip()
+            if not command:
+                continue
+            step = record.get("step", "?")
+            cmd_name = str(record.get("cmd_name", "")).strip() or _extract_command_name(command)
+            target = str(record.get("target", "")).strip()
+            if target:
+                synthetic.append(f"[synthetic] step={step} cmd={cmd_name} target={target} :: {command}")
+            else:
+                synthetic.append(f"[synthetic] step={step} cmd={cmd_name} :: {command}")
+        return synthetic[-20:]
 
     # -----------------------------------------------------------------
     # Introspection (for reward computation and debugging)

src/open_range/server/runtime.py

@@ -68,6 +68,13 @@ _VALIDATOR_PROFILE_ALIASES = {
 }
 _LIVE_VALIDATOR_PROFILES = {"training"}
 _ALLOW_OFFLINE_ADMISSION_ENV = "OPENRANGE_ALLOW_OFFLINE_ADMISSION"
+_PERSISTED_SNAPSHOT_VALIDATION_ALIASES = {
+    "none": "trust",
+    "disabled": "trust",
+    "off": "trust",
+    "revalidate": "offline",
+    "strict": "offline",
+}
 
 
 def _env_flag(name: str, default: bool = False) -> bool:
@@ -320,6 +327,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),
@@ -395,6 +413,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 = (
@@ -418,7 +437,16 @@ class ManagedSnapshotRuntime:
             or os.getenv("OPENRANGE_RUNTIME_VALIDATOR_PROFILE", _DEFAULT_VALIDATOR_PROFILE)
         )
         self._enforce_validator_profile_policy()
+        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)
@@ -475,6 +503,10 @@ class ManagedSnapshotRuntime:
                 "OPENRANGE_ENABLE_PATCH_VALIDATION",
                 default=False,
             ),
+            persisted_snapshot_validation=os.getenv(
+                "OPENRANGE_PERSISTED_SNAPSHOT_VALIDATION",
+                "offline",
+            ),
         )
 
     def _enforce_validator_profile_policy(self) -> None:
@@ -516,6 +548,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:
@@ -567,6 +600,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
@@ -583,14 +617,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()
@@ -608,32 +643,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 {
@@ -644,6 +676,7 @@ class ManagedSnapshotRuntime:
             "parent_selection_strategy": self.parent_selection_strategy,
             "validator_profile": self.validator_profile,
             "allow_insecure_offline_profile": self.allow_insecure_offline_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(),
@@ -706,30 +739,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_console_context.py

@@ -0,0 +1,64 @@
+"""Unit tests for console environment context resolution."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from open_range.server.console import _get_env_context
+from open_range.server.environment import RangeEnvironment
+
+
+class _Req:
+    def __init__(self, app):
+        self.app = app
+
+
+def _app_with_state(**kwargs):
+    return SimpleNamespace(state=SimpleNamespace(**kwargs))
+
+
+def test_prefers_active_websocket_session_env():
+    fallback_env = RangeEnvironment(docker_available=False)
+    ws_env = RangeEnvironment(docker_available=False)
+    server = SimpleNamespace(
+        _sessions={"session_a": ws_env},
+        _session_info={"session_a": SimpleNamespace(last_activity_at=10.0)},
+    )
+    request = _Req(_app_with_state(env=fallback_env, openenv_server=server))
+
+    ctx = _get_env_context(request)
+    assert ctx["env"] is ws_env
+    assert ctx["state_scope"] == "websocket_session"
+    assert ctx["session_id"] == "session_a"
+    assert ctx["warning"] is None
+
+
+def test_uses_app_state_env_when_no_active_session():
+    fallback_env = RangeEnvironment(docker_available=False)
+    server = SimpleNamespace(_sessions={}, _session_info={})
+    request = _Req(_app_with_state(env=fallback_env, openenv_server=server))
+
+    ctx = _get_env_context(request)
+    assert ctx["env"] is fallback_env
+    assert ctx["state_scope"] == "app_state_env"
+    assert ctx["session_id"] is None
+    assert isinstance(ctx["warning"], str) and ctx["warning"]
+
+
+def test_multiple_sessions_selects_most_recent_and_warns():
+    older_env = RangeEnvironment(docker_available=False)
+    newer_env = RangeEnvironment(docker_available=False)
+    server = SimpleNamespace(
+        _sessions={"old": older_env, "new": newer_env},
+        _session_info={
+            "old": SimpleNamespace(last_activity_at=10.0),
+            "new": SimpleNamespace(last_activity_at=20.0),
+        },
+    )
+    request = _Req(_app_with_state(openenv_server=server))
+
+    ctx = _get_env_context(request)
+    assert ctx["env"] is newer_env
+    assert ctx["state_scope"] == "websocket_session"
+    assert ctx["session_id"] == "new"
+    assert "active sessions" in (ctx["warning"] or "").lower()

tests/test_environment.py

@@ -66,6 +66,12 @@ class TestReset:
         assert isinstance(obs, RangeObservation)
         assert env.snapshot is not None
 
+    def test_reset_initializes_services_status_from_topology_hosts(self):
+        env = RangeEnvironment(docker_available=False)
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
+        # In mock mode service health is unknown, but hosts should be tracked.
+        assert set(env.state.services_status.keys()) == {"attacker", "siem"}
+
 
 class TestTargetResolution:
     """Target selection should honor manifest-compiled metadata."""
@@ -187,6 +193,14 @@ class TestBlueStep:
         obs = env.step(RangeAction(command="", mode="blue"))
         assert obs.stderr != ""
 
+    def test_blue_alerts_fall_back_to_synthetic_red_history(self):
+        env = RangeEnvironment(docker_available=False)
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
+        env.step(RangeAction(command="nmap -sV web", mode="red"))
+        obs = env.step(RangeAction(command="tail -n 50 /var/log/siem/all.log", mode="blue"))
+        assert obs.alerts
+        assert any("synthetic" in alert.lower() for alert in obs.alerts)
+
     def test_step_passes_timeout_override_to_executor(self):
         env = RangeEnvironment(docker_available=False)
         env.reset(snapshot=_MINIMAL_SNAPSHOT)

tests/test_runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 import pytest
@@ -120,6 +121,50 @@ 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,
+            validator_profile="offline",
+            allow_insecure_offline_profile=True,
+            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,
+            validator_profile="offline",
+            allow_insecure_offline_profile=True,
+            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,
+            validator_profile="offline",
+            allow_insecure_offline_profile=True,
+            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,