Merge pull request #52 from open-cybernauts/feat/issue50-canonical-base-graph-policy

README.md

@@ -20,14 +20,14 @@ A multi-agent cybersecurity gymnasium on [OpenEnv](https://github.com/meta-pytor
 
 ## How It Works
 
-A **manifest** declares a family of legal enterprise worlds — topology, services, identities, trust relationships, vulnerability classes, and mutation bounds. A shared **ManagedSnapshotRuntime** inside the shipped OpenEnv server process owns the admitted snapshot population. It compiles a root snapshot from the manifest, then derives child snapshots by applying explicit typed mutations to admitted parents. Each candidate child is structurally validated, rendered into a concrete artifact bundle under `snapshots/<id>/`, and, in managed-generation mode, can also be booted and live-validated before admission. `reset()` selects one frozen admitted snapshot. `step()` runs commands inside it.
+A **manifest** declares a family of legal enterprise worlds — topology, services, identities, trust relationships, vulnerability classes, and mutation bounds. A shared **ManagedSnapshotRuntime** inside the shipped OpenEnv server process owns the admitted snapshot population. It compiles a graph-friendly root snapshot from the manifest, normalizing trust-only principals into a canonical principal catalog, then derives child snapshots by applying explicit typed mutations to admitted parents. Parent selection is policy-driven over the admitted population rather than raw latest/random sampling. Each candidate child is validated in layers: manifest compliance, canonical graph checks, structural/task checks, and, in managed-generation mode, booted runtime checks before admission. `reset()` selects one frozen admitted snapshot. `step()` runs commands inside it.
 
 ```mermaid
 flowchart LR
     M[Manifest<br/>legal family +<br/>mutation envelope] --> B[Base snapshot compiler]
     B --> P[Admitted root snapshot]
     P --> R[ManagedSnapshotRuntime<br/>shared inside server process]
-    R --> U[Parent selector +<br/>typed mutator]
+    R --> U[Policy-guided parent selector +<br/>typed mutator]
     U --> V{Validator<br/>manifest + graph +<br/>runtime checks}
     V -->|fail| U
     V -->|pass| S[Admitted snapshot population]
@@ -85,13 +85,13 @@ uv run pytest tests/ -v --tb=short
 
 **Manifest** — YAML defining the legal world family and mutation envelope: hosts, zones, services, users, NPCs, data assets, credential policies, monitoring coverage, trust relationships, and which vulnerability classes the runtime may plant or extend. Three example manifests ship (healthcare, fintech, SaaS) at tiers 1-3.
 
-**ManagedSnapshotRuntime** — Shared singleton created at server startup. Owns the `SnapshotStore`, base builder, parent-snapshot mutator, validator gate, `SnapshotRenderer`, snapshot preload, optional background refill, and episode-result feedback. This is the hidden orchestrator behind the env; callers still only see `reset()`, `step()`, and `state()`.
+**ManagedSnapshotRuntime** — Shared singleton created at server startup. Owns the `SnapshotStore`, base builder, population-aware parent selector, parent-snapshot mutator, validator gate, `SnapshotRenderer`, snapshot preload, optional background refill, and episode-result feedback. This is the hidden orchestrator behind the env; callers still only see `reset()`, `step()`, and `state()`.
 
-**Builder / Mutator** — The base builder compiles an initial `SnapshotSpec` from a manifest. The mutator then derives child `SnapshotSpec`s from admitted parents using typed mutation plans plus curriculum context. Each snapshot carries lineage metadata (`snapshot_id`, `parent_snapshot_id`, `root_snapshot_id`, generation depth, mutation summary) and can emit constrained service/app payloads through `SnapshotSpec.files`. Three base builders ship: `LLMSnapshotBuilder` (production, via litellm), `TemplateOnlyBuilder` (deterministic shipped default), `FileBuilder` (load from disk).
+**Builder / Mutator** — The base builder compiles an initial `SnapshotSpec` from a manifest. Root hydration then expands that into canonical topology state: host details, dependency edges, trust edges, and a principal catalog that can represent trust-only people without inventing login accounts. The mutator derives child `SnapshotSpec`s from admitted parents using typed mutation plans plus an explicit mutation-policy layer that scores parents and candidate edits with curriculum, replay, novelty, and lineage signals. Each snapshot carries lineage metadata (`snapshot_id`, `parent_snapshot_id`, `root_snapshot_id`, generation depth, mutation summary) and can emit constrained service/app payloads through `SnapshotSpec.files`. Three base builders ship: `LLMSnapshotBuilder` (production, via litellm), `TemplateOnlyBuilder` (deterministic shipped default), `FileBuilder` (load from disk).
 
 The deployed package exposes the standard OpenEnv `reset()`, `step()`, and `state()` contract through `server.app:app`, which is the entrypoint referenced by `openenv.yaml`.
 
-**Validator** — Admission gate for candidate snapshots. The shipped runtime enforces manifest compliance and graph consistency before structural/task checks. When `OPENRANGE_ENABLE_LIVE_ADMISSION=1`, the runtime also boots the rendered child bundle, applies rendered payload files, constructs a real `ContainerSet`, and runs live build/exploit/evidence/reward checks before admission. Public/HF mode can still rely on a prebuilt admitted pool with live admission disabled.
+**Validator** — Admission gate for candidate snapshots. The shipped runtime first enforces manifest compliance plus graph-native checks such as graph consistency, path solvability, evidence sufficiency, and reward grounding before structural/task checks. When `OPENRANGE_ENABLE_LIVE_ADMISSION=1`, the runtime also boots the rendered child bundle, applies rendered payload files, constructs a real `ContainerSet`, and runs live build/exploit/evidence/reward checks before admission. Public/HF mode can still rely on a prebuilt admitted pool with live admission disabled.
 
 **Environment** — `RangeEnvironment(Environment)` following the OpenEnv contract. `reset()` asks the shared runtime for a frozen admitted snapshot. `step(action)` routes commands to the appropriate container — Red runs on the attacker box, Blue runs on the SIEM. No artificial command allowlists; the container's installed tools are the constraint.
 

src/open_range/builder/manifest_graph.py

@@ -0,0 +1,340 @@
+"""Manifest-to-topology compilation helpers for root snapshot hydration.
+
+These helpers turn a manifest's declared company world into the canonical
+topology fields the mutator, validators, and runtime expect to reason about.
+They intentionally keep "real login users" separate from trust-only narrative
+principals so the trust graph can be compiled without silently creating extra
+accounts in rendered services.
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+
+def build_host_catalog(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    """Return the manifest-defined host catalog keyed by host name."""
+    catalog: dict[str, dict[str, Any]] = {}
+    for raw in manifest.get("topology", {}).get("hosts", []):
+        if not isinstance(raw, dict):
+            continue
+        name = str(raw.get("name", "")).strip()
+        if not name:
+            continue
+        catalog[name] = {
+            "zone": str(raw.get("zone", "")),
+            "services": deepcopy(raw.get("services", [])),
+            "connects_to": deepcopy(raw.get("connects_to", [])),
+            "purpose": str(raw.get("purpose", "")),
+            "hostname": str(raw.get("hostname", "")),
+            "os": str(raw.get("os", "")),
+            "exposure": deepcopy(raw.get("exposure", {})),
+        }
+    return catalog
+
+
+def build_principal_catalog(
+    manifest: dict[str, Any],
+    existing: dict[str, Any] | None = None,
+) -> tuple[dict[str, dict[str, Any]], list[str]]:
+    """Return a canonical principal catalog plus normalized trust-only names."""
+    catalog: dict[str, dict[str, Any]] = {}
+    trust_only: set[str] = set()
+
+    if isinstance(existing, dict):
+        for name, raw in existing.items():
+            principal = str(name).strip()
+            if not principal or not isinstance(raw, dict):
+                continue
+            catalog[principal] = deepcopy(raw)
+
+    for raw in manifest.get("users", []):
+        if not isinstance(raw, dict):
+            continue
+        username = str(raw.get("username", "")).strip()
+        if not username:
+            continue
+        principal = catalog.setdefault(username, {})
+        principal.update(
+            {
+                "username": username,
+                "kind": "user",
+                "is_login_account": True,
+                "hosts": deepcopy(raw.get("hosts", [])),
+                "department": str(raw.get("department", "")),
+                "role": str(raw.get("role", "")),
+                "email": str(raw.get("email", "")),
+                "full_name": str(raw.get("full_name", "")),
+            }
+        )
+
+    for raw in manifest.get("trust_relationships", []):
+        if not isinstance(raw, dict):
+            continue
+        source = str(raw.get("source") or raw.get("from") or "").strip()
+        target = str(raw.get("target") or raw.get("to") or "").strip()
+        for principal_name in (source, target):
+            if not principal_name:
+                continue
+            principal = catalog.setdefault(principal_name, {})
+            if not principal.get("is_login_account", False):
+                trust_only.add(principal_name)
+            principal.setdefault("username", principal_name)
+            principal.setdefault("kind", "trust_principal")
+            principal.setdefault("is_login_account", False)
+            principal.setdefault("hosts", [])
+            principal.setdefault("department", "")
+            principal.setdefault("role", "")
+            principal.setdefault("email", "")
+            principal.setdefault("full_name", "")
+
+    return catalog, sorted(trust_only)
+
+
+def compile_manifest_topology(
+    manifest: dict[str, Any],
+    topology: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Compile manifest state into graph-friendly topology fields.
+
+    Existing topology fields are preserved where possible so builder-generated
+    details such as passwords or payload-specific knobs survive root hydration.
+    """
+    compiled = deepcopy(topology) if isinstance(topology, dict) else {}
+    company = manifest.get("company", {}) if isinstance(manifest.get("company"), dict) else {}
+
+    compiled.setdefault("tier", int(manifest.get("tier", compiled.get("tier", 1)) or 1))
+    compiled.setdefault("domain", company.get("domain", "acmecorp.local"))
+    compiled.setdefault("org_name", company.get("name", "AcmeCorp"))
+    compiled.setdefault("manifest_name", manifest.get("name", ""))
+    compiled.setdefault("difficulty", deepcopy(manifest.get("difficulty", {})))
+    compiled.setdefault(
+        "networks",
+        deepcopy(manifest.get("topology", {}).get("networks", [])),
+    )
+    compiled.setdefault(
+        "firewall_rules",
+        deepcopy(manifest.get("topology", {}).get("firewall_rules", [])),
+    )
+
+    host_catalog = build_host_catalog(manifest)
+    compiled["host_catalog"] = host_catalog
+    compiled["hosts"] = _merge_hosts(compiled.get("hosts"), host_catalog)
+    compiled["zones"] = _merge_zones(compiled.get("zones"), host_catalog)
+    compiled["users"] = _merge_users(compiled.get("users"), manifest)
+    compiled["host_details"] = _merge_host_details(compiled.get("host_details"), host_catalog)
+    compiled["dependency_edges"] = _merge_dependency_edges(
+        compiled.get("dependency_edges"),
+        host_catalog,
+    )
+
+    principal_catalog, trust_only = build_principal_catalog(
+        manifest,
+        existing=compiled.get("principal_catalog")
+        if isinstance(compiled.get("principal_catalog"), dict)
+        else None,
+    )
+    compiled["principal_catalog"] = principal_catalog
+    compiled["trust_edges"] = _merge_trust_edges(compiled.get("trust_edges"), manifest)
+    compiled["manifest_normalization"] = {
+        "trust_only_principals": trust_only,
+        "notes": [
+            (
+                "Normalized trust principals not present in manifest users into "
+                "principal_catalog only"
+            )
+        ]
+        if trust_only
+        else [],
+    }
+    return compiled
+
+
+def _merge_hosts(
+    raw_hosts: object,
+    host_catalog: dict[str, dict[str, Any]],
+) -> list[str]:
+    hosts: list[str] = []
+    seen: set[str] = set()
+    if isinstance(raw_hosts, list):
+        for raw in raw_hosts:
+            if isinstance(raw, dict):
+                name = str(raw.get("name", "")).strip()
+            else:
+                name = str(raw).strip()
+            if not name or name in seen:
+                continue
+            seen.add(name)
+            hosts.append(name)
+    for host in host_catalog:
+        if host in seen:
+            continue
+        seen.add(host)
+        hosts.append(host)
+    return hosts
+
+
+def _merge_zones(
+    raw_zones: object,
+    host_catalog: dict[str, dict[str, Any]],
+) -> dict[str, list[str]]:
+    zones: dict[str, list[str]] = {}
+    if isinstance(raw_zones, dict):
+        for zone, raw_hosts in raw_zones.items():
+            zone_name = str(zone).strip()
+            if not zone_name:
+                continue
+            zone_hosts: list[str] = []
+            if isinstance(raw_hosts, list):
+                for raw_host in raw_hosts:
+                    host = str(raw_host).strip()
+                    if host and host not in zone_hosts:
+                        zone_hosts.append(host)
+            zones[zone_name] = zone_hosts
+
+    for host, raw_catalog in host_catalog.items():
+        zone = str(raw_catalog.get("zone", "")).strip() or "default"
+        zone_hosts = zones.setdefault(zone, [])
+        if host not in zone_hosts:
+            zone_hosts.append(host)
+    return zones
+
+
+def _merge_users(raw_users: object, manifest: dict[str, Any]) -> list[dict[str, Any]]:
+    existing: dict[str, dict[str, Any]] = {}
+    extras: list[dict[str, Any]] = []
+    if isinstance(raw_users, list):
+        for raw in raw_users:
+            if not isinstance(raw, dict):
+                continue
+            username = str(raw.get("username", "")).strip()
+            if not username:
+                continue
+            existing[username] = deepcopy(raw)
+
+    merged: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    for raw in manifest.get("users", []):
+        if not isinstance(raw, dict):
+            continue
+        username = str(raw.get("username", "")).strip()
+        if not username:
+            continue
+        record = existing.pop(username, {})
+        record.setdefault("username", username)
+        record.setdefault("password", "")
+        record.setdefault("groups", [])
+        record.setdefault("hosts", deepcopy(raw.get("hosts", [])))
+        record.setdefault("email", str(raw.get("email", "")))
+        record.setdefault("full_name", str(raw.get("full_name", "")))
+        record.setdefault("department", str(raw.get("department", "")))
+        record.setdefault("role", str(raw.get("role", "")))
+        merged.append(record)
+        seen.add(username)
+
+    for username, record in existing.items():
+        if username in seen:
+            continue
+        extras.append(record)
+    merged.extend(extras)
+    return merged
+
+
+def _merge_host_details(
+    raw_details: object,
+    host_catalog: dict[str, dict[str, Any]],
+) -> dict[str, dict[str, Any]]:
+    host_details: dict[str, dict[str, Any]] = {}
+    if isinstance(raw_details, dict):
+        for host, raw_detail in raw_details.items():
+            host_name = str(host).strip()
+            if not host_name or not isinstance(raw_detail, dict):
+                continue
+            host_details[host_name] = deepcopy(raw_detail)
+
+    for host, raw_catalog in host_catalog.items():
+        detail = host_details.setdefault(host, {})
+        detail.setdefault("zone", str(raw_catalog.get("zone", "")))
+        detail.setdefault("services", deepcopy(raw_catalog.get("services", [])))
+        detail.setdefault("connects_to", deepcopy(raw_catalog.get("connects_to", [])))
+        detail.setdefault("purpose", str(raw_catalog.get("purpose", "")))
+        detail.setdefault("hostname", str(raw_catalog.get("hostname", "")))
+        detail.setdefault("os", str(raw_catalog.get("os", "")))
+        detail.setdefault("exposure", deepcopy(raw_catalog.get("exposure", {})))
+    return host_details
+
+
+def _merge_dependency_edges(
+    raw_edges: object,
+    host_catalog: dict[str, dict[str, Any]],
+) -> list[dict[str, str]]:
+    edges: list[dict[str, str]] = []
+    seen: set[tuple[str, str]] = set()
+    if isinstance(raw_edges, list):
+        for raw in raw_edges:
+            if not isinstance(raw, dict):
+                continue
+            source = str(raw.get("source", "")).strip()
+            target = str(raw.get("target", "")).strip()
+            if not source or not target or (source, target) in seen:
+                continue
+            edges.append({"source": source, "target": target})
+            seen.add((source, target))
+
+    for source, raw_catalog in host_catalog.items():
+        raw_targets = raw_catalog.get("connects_to", [])
+        if not isinstance(raw_targets, list):
+            continue
+        for raw_target in raw_targets:
+            target = str(raw_target).strip()
+            if not target or (source, target) in seen:
+                continue
+            edges.append({"source": source, "target": target})
+            seen.add((source, target))
+    return edges
+
+
+def _merge_trust_edges(
+    raw_edges: object,
+    manifest: dict[str, Any],
+) -> list[dict[str, str]]:
+    edges: list[dict[str, str]] = []
+    seen: set[tuple[str, str, str]] = set()
+    if isinstance(raw_edges, list):
+        for raw in raw_edges:
+            if not isinstance(raw, dict):
+                continue
+            source = str(raw.get("source", "")).strip()
+            target = str(raw.get("target", "")).strip()
+            edge_type = str(raw.get("type", "")).strip()
+            if not source or not target or (source, target, edge_type) in seen:
+                continue
+            edges.append(
+                {
+                    "source": source,
+                    "target": target,
+                    "type": edge_type,
+                    "context": str(raw.get("context", "")),
+                }
+            )
+            seen.add((source, target, edge_type))
+
+    for raw in manifest.get("trust_relationships", []):
+        if not isinstance(raw, dict):
+            continue
+        source = str(raw.get("source") or raw.get("from") or "").strip()
+        target = str(raw.get("target") or raw.get("to") or "").strip()
+        edge_type = str(raw.get("type", "")).strip()
+        if not source or not target or (source, target, edge_type) in seen:
+            continue
+        edges.append(
+            {
+                "source": source,
+                "target": target,
+                "type": edge_type,
+                "context": str(raw.get("context") or raw.get("description") or ""),
+            }
+        )
+        seen.add((source, target, edge_type))
+    return edges

src/open_range/builder/mutation_policy.py

@@ -0,0 +1,328 @@
+"""Population-aware parent and mutation selection policy."""
+
+from __future__ import annotations
+
+import random
+from collections import Counter
+from dataclasses import dataclass
+from typing import Any
+
+from open_range.protocols import BuildContext, MutationOp, SnapshotSpec
+from open_range.validator.graphs import compile_snapshot_graphs
+
+
+@dataclass(frozen=True, slots=True)
+class ParentPolicyScore:
+    snapshot_id: str
+    total: float
+    components: dict[str, float]
+
+
+@dataclass(frozen=True, slots=True)
+class MutationChoice:
+    op: MutationOp
+    total: float
+    components: dict[str, float]
+
+
+class PopulationMutationPolicy:
+    """Simple population-guided policy for parent and op selection.
+
+    This is intentionally heuristic rather than learned. It gives the runtime
+    an explicit place to score parents and mutation candidates using curriculum,
+    replay, novelty, and lineage signals instead of relying on raw RNG.
+    """
+
+    name = "population_guided_v1"
+
+    def select_parent(
+        self,
+        entries: list[Any],
+        *,
+        context: BuildContext,
+        snapshot_stats: dict[str, dict[str, Any]],
+        rng: random.Random,
+    ) -> tuple[Any, ParentPolicyScore]:
+        scores = self.score_parents(
+            entries,
+            context=context,
+            snapshot_stats=snapshot_stats,
+        )
+        if not scores:
+            raise ValueError("No parent candidates available")
+        ordered = sorted(scores, key=lambda score: score.total, reverse=True)
+        top = ordered[: min(3, len(ordered))]
+        weights = [max(score.total, 0.05) for score in top]
+        chosen_score = rng.choices(top, weights=weights, k=1)[0]
+        chosen_entry = next(
+            entry for entry in entries if entry.snapshot_id == chosen_score.snapshot_id
+        )
+        return chosen_entry, chosen_score
+
+    def score_parents(
+        self,
+        entries: list[Any],
+        *,
+        context: BuildContext,
+        snapshot_stats: dict[str, dict[str, Any]],
+    ) -> list[ParentPolicyScore]:
+        if not entries:
+            return []
+
+        root_counts = Counter(
+            entry.snapshot.lineage.root_snapshot_id or entry.snapshot_id
+            for entry in entries
+        )
+        vuln_frequency = Counter()
+        for entry in entries:
+            vuln_frequency.update(v.type for v in entry.snapshot.truth_graph.vulns if v.type)
+
+        scores: list[ParentPolicyScore] = []
+        for entry in entries:
+            snapshot = entry.snapshot
+            stat = snapshot_stats.get(entry.snapshot_id, {})
+            vuln_types = {v.type for v in snapshot.truth_graph.vulns if v.type}
+            compiled = compile_snapshot_graphs(snapshot)
+
+            plays = float(stat.get("plays", 0))
+            red_rate = float(stat.get("red_solve_rate", 0.0))
+            blue_rate = float(stat.get("blue_detect_rate", 0.0))
+            frontier = (
+                0.4
+                if plays == 0
+                else (
+                    self._frontier_score(red_rate)
+                    + self._frontier_score(blue_rate)
+                )
+                / 2.0
+            )
+            replay = 1.0 / (plays + 1.0)
+            novelty = 1.0 / (
+                1.0 + sum(vuln_frequency[vuln] for vuln in vuln_types)
+            ) if vuln_types else 0.25
+            weak_overlap = float(len(vuln_types.intersection(context.weak_areas)))
+            root_id = snapshot.lineage.root_snapshot_id or entry.snapshot_id
+            lineage_balance = 1.0 / max(root_counts[root_id], 1)
+            depth = float(snapshot.lineage.generation_depth)
+            depth_balance = 1.0 / (1.0 + max(depth - 3.0, 0.0))
+            recency = 1.0 / (1.0 + float(stat.get("plays_recent", 0)))
+            complexity = min(
+                (
+                    len(snapshot.truth_graph.vulns) * 0.25
+                    + len(snapshot.golden_path) * 0.03
+                    + len(compiled.dependency_edges) * 0.02
+                    + len(compiled.trust_edges) * 0.02
+                ),
+                1.0,
+            )
+
+            components = {
+                "frontier": frontier,
+                "replay": replay,
+                "novelty": novelty,
+                "weak_overlap": weak_overlap,
+                "lineage_balance": lineage_balance,
+                "depth_balance": depth_balance,
+                "recency": recency,
+                "complexity": complexity,
+            }
+            total = (
+                frontier * 0.28
+                + replay * 0.18
+                + novelty * 0.16
+                + weak_overlap * 0.18
+                + lineage_balance * 0.08
+                + depth_balance * 0.04
+                + recency * 0.04
+                + complexity * 0.04
+            )
+            scores.append(
+                ParentPolicyScore(
+                    snapshot_id=entry.snapshot_id,
+                    total=round(max(total, 0.05), 4),
+                    components={key: round(value, 4) for key, value in components.items()},
+                )
+            )
+        return scores
+
+    def choose_mutations(
+        self,
+        *,
+        structural_candidates: list[MutationOp],
+        security_candidates: list[MutationOp],
+        snapshot: SnapshotSpec,
+        context: BuildContext,
+        rng: random.Random,
+    ) -> tuple[list[MutationOp], float, dict[str, float]]:
+        selected: list[MutationChoice] = []
+
+        structural = self._select_candidate(
+            structural_candidates,
+            snapshot=snapshot,
+            context=context,
+            rng=rng,
+        )
+        if structural is not None:
+            selected.append(structural)
+
+        security_pool = [
+            choice
+            for choice in (
+                self._select_candidate(
+                    security_candidates,
+                    snapshot=snapshot,
+                    context=context,
+                    rng=rng,
+                ),
+            )
+            if choice is not None
+        ]
+        selected.extend(security_pool)
+
+        if not selected and security_candidates:
+            fallback = self._select_candidate(
+                security_candidates,
+                snapshot=snapshot,
+                context=context,
+                rng=rng,
+                deterministic=True,
+            )
+            if fallback is not None:
+                selected.append(fallback)
+
+        if not structural and len(security_candidates) > 1:
+            ranked = self._rank_candidates(
+                security_candidates,
+                snapshot=snapshot,
+                context=context,
+            )
+            for choice in ranked:
+                if any(choice.op.mutation_id == existing.op.mutation_id for existing in selected):
+                    continue
+                selected.append(choice)
+                break
+
+        ops = [choice.op for choice in selected]
+        if not ops:
+            return [], 0.0, {}
+
+        breakdown = {
+            "curriculum": round(sum(c.components["curriculum"] for c in selected), 4),
+            "novelty": round(sum(c.components["novelty"] for c in selected), 4),
+            "structural_gain": round(sum(c.components["structural_gain"] for c in selected), 4),
+            "lineage": round(sum(c.components["lineage"] for c in selected), 4),
+        }
+        total = round(sum(choice.total for choice in selected), 4)
+        return ops, total, breakdown
+
+    def _select_candidate(
+        self,
+        candidates: list[MutationOp],
+        *,
+        snapshot: SnapshotSpec,
+        context: BuildContext,
+        rng: random.Random,
+        deterministic: bool = False,
+    ) -> MutationChoice | None:
+        ranked = self._rank_candidates(
+            candidates,
+            snapshot=snapshot,
+            context=context,
+        )
+        if not ranked:
+            return None
+        if deterministic or len(ranked) == 1:
+            return ranked[0]
+        top = ranked[: min(3, len(ranked))]
+        weights = [max(choice.total, 0.05) for choice in top]
+        return rng.choices(top, weights=weights, k=1)[0]
+
+    def _rank_candidates(
+        self,
+        candidates: list[MutationOp],
+        *,
+        snapshot: SnapshotSpec,
+        context: BuildContext,
+    ) -> list[MutationChoice]:
+        ranked: list[MutationChoice] = []
+        existing_vulns = {v.type for v in snapshot.truth_graph.vulns if v.type}
+        for candidate in candidates:
+            curriculum = self._curriculum_bonus(candidate, context, existing_vulns)
+            novelty = self._novelty_bonus(candidate, context)
+            structural_gain = self._structural_gain(candidate)
+            lineage = 1.0 / (1.0 + snapshot.lineage.generation_depth)
+            components = {
+                "curriculum": curriculum,
+                "novelty": novelty,
+                "structural_gain": structural_gain,
+                "lineage": lineage,
+            }
+            total = (
+                curriculum * 0.38
+                + novelty * 0.24
+                + structural_gain * 0.28
+                + lineage * 0.10
+            )
+            ranked.append(
+                MutationChoice(
+                    op=candidate,
+                    total=round(max(total, 0.05), 4),
+                    components={key: round(value, 4) for key, value in components.items()},
+                )
+            )
+        ranked.sort(key=lambda choice: choice.total, reverse=True)
+        return ranked
+
+    @staticmethod
+    def _frontier_score(rate: float) -> float:
+        return max(0.0, 1.0 - abs(rate - 0.5) * 2.0)
+
+    @staticmethod
+    def _structural_gain(op: MutationOp) -> float:
+        mapping = {
+            "add_service": 1.0,
+            "add_dependency_edge": 0.9,
+            "add_trust_edge": 0.85,
+            "add_user": 0.8,
+            "seed_vuln": 0.7,
+            "add_benign_noise": 0.3,
+        }
+        return mapping.get(op.op_type, 0.2) * max(op.magnitude, 1)
+
+    @staticmethod
+    def _novelty_bonus(op: MutationOp, context: BuildContext) -> float:
+        bonus = 0.4
+        if op.op_type == "seed_vuln":
+            vuln_type = str(op.params.get("vuln_type", "")).strip()
+            if vuln_type and vuln_type not in context.previous_vuln_classes:
+                bonus += 1.0
+        if op.op_type == "add_benign_noise":
+            location = str(op.params.get("location", "")).strip()
+            if location and location not in context.recent_attack_surfaces:
+                bonus += 0.5
+        if op.op_type not in {"seed_vuln", "add_benign_noise"}:
+            bonus += 0.4
+        return bonus
+
+    @staticmethod
+    def _curriculum_bonus(
+        op: MutationOp,
+        context: BuildContext,
+        existing_vulns: set[str],
+    ) -> float:
+        bonus = 0.35
+        if op.op_type == "seed_vuln":
+            vuln_type = str(op.params.get("vuln_type", "")).strip()
+            if vuln_type in context.weak_areas:
+                bonus += 1.5
+            if vuln_type and vuln_type not in existing_vulns:
+                bonus += 0.4
+        if op.op_type in {"add_dependency_edge", "add_trust_edge"} and context.require_chain_length > 1:
+            bonus += 0.6
+        if context.focus_layer == "identity" and op.op_type in {"add_user", "add_trust_edge"}:
+            bonus += 0.5
+        if context.focus_layer == "infra" and op.op_type in {"add_service", "add_dependency_edge"}:
+            bonus += 0.5
+        if context.focus_layer == "process" and op.op_type == "add_benign_noise":
+            bonus += 0.4
+        return bonus

src/open_range/builder/mutator.py

@@ -14,6 +14,8 @@ from copy import deepcopy
 from typing import Any
 
 from open_range.builder.builder import render_template_payloads
+from open_range.builder.manifest_graph import compile_manifest_topology
+from open_range.builder.mutation_policy import PopulationMutationPolicy
 from open_range.protocols import (
     BuildContext,
     EvidenceItem,
@@ -49,6 +51,7 @@ class Mutator:
         self,
         builder: SnapshotBuilder,
         max_retries: int = 3,
+        policy: PopulationMutationPolicy | None = None,
     ) -> None:
         """Initialize the mutator with a builder and retry limit.
 
@@ -58,6 +61,7 @@ class Mutator:
         """
         self.builder = builder
         self.max_retries = max_retries
+        self.policy = policy or PopulationMutationPolicy()
         self._history: list[str] = []  # recent vuln classes
         self._attack_surfaces: list[str] = []  # recent injection points
         self._episode_count: int = 0
@@ -230,23 +234,19 @@ class Mutator:
         manifest: dict[str, Any],
     ) -> SnapshotSpec:
         root = snapshot.model_copy(deep=True)
-        topology = dict(root.topology)
-        company = manifest.get("company", {}) if isinstance(manifest.get("company"), dict) else {}
-        topology.setdefault("domain", company.get("domain", "acmecorp.local"))
-        topology.setdefault("org_name", company.get("name", "AcmeCorp"))
-        topology.setdefault("manifest_name", manifest.get("name", ""))
-        topology.setdefault("difficulty", deepcopy(manifest.get("difficulty", {})))
-        topology.setdefault("host_catalog", _build_host_catalog(manifest))
-        topology.setdefault("host_details", {})
-        topology.setdefault("dependency_edges", [])
-        topology.setdefault("trust_edges", [])
-        root.topology = topology
+        root.topology = compile_manifest_topology(manifest, root.topology)
         root.lineage = LineageMetadata(
             manifest_id=str(manifest.get("name", "")),
             generation_depth=0,
             mutation_summary=["compile_base_snapshot"],
         )
         root.mutation_plan = None
+        normalization = root.topology.get("manifest_normalization", {})
+        if isinstance(normalization, dict):
+            notes = normalization.get("notes", [])
+            if isinstance(notes, list):
+                for note in notes:
+                    logger.info("Mutator: manifest normalization applied: %s", note)
         return root
 
     def _mutate_parent_snapshot(
@@ -293,7 +293,6 @@ class Mutator:
         rng: random.Random,
     ) -> MutationPlan:
         ops: list[MutationOp] = []
-        used_ids: set[str] = set()
 
         structural_candidates = []
         op = self._candidate_add_service(manifest, snapshot, rng)
@@ -309,11 +308,6 @@ class Mutator:
         if op is not None:
             structural_candidates.append(op)
 
-        if structural_candidates:
-            chosen = rng.choice(structural_candidates)
-            ops.append(chosen)
-            used_ids.add(chosen.mutation_id)
-
         security_candidates = []
         op = self._candidate_seed_vuln(manifest, snapshot, context, rng)
         if op is not None:
@@ -322,10 +316,13 @@ class Mutator:
         if op is not None:
             security_candidates.append(op)
 
-        if security_candidates:
-            chosen = rng.choice(security_candidates)
-            if chosen.mutation_id not in used_ids:
-                ops.append(chosen)
+        ops, policy_score, score_breakdown = self.policy.choose_mutations(
+            structural_candidates=structural_candidates,
+            security_candidates=security_candidates,
+            snapshot=snapshot,
+            context=context,
+            rng=rng,
+        )
 
         if not ops:
             fallback = self._candidate_add_benign_noise(snapshot, rng)
@@ -338,6 +335,9 @@ class Mutator:
             predicted_complexity_delta=len(ops),
             predicted_chain_delta=sum(1 for op in ops if op.op_type == "seed_vuln"),
             predicted_novelty=round(0.2 * len({op.op_type for op in ops}), 2),
+            policy_name=self.policy.name,
+            policy_score=policy_score,
+            score_breakdown=score_breakdown,
         )
 
     def _candidate_add_service(
@@ -559,6 +559,7 @@ class Mutator:
         host_details = topology.setdefault("host_details", {})
         dependency_edges = topology.setdefault("dependency_edges", [])
         trust_edges = topology.setdefault("trust_edges", [])
+        principal_catalog = topology.setdefault("principal_catalog", {})
         users = topology.setdefault("users", [])
 
         if not isinstance(host_details, dict):
@@ -570,6 +571,9 @@ class Mutator:
         if not isinstance(trust_edges, list):
             trust_edges = []
             topology["trust_edges"] = trust_edges
+        if not isinstance(principal_catalog, dict):
+            principal_catalog = {}
+            topology["principal_catalog"] = principal_catalog
         if not isinstance(users, list):
             users = []
             topology["users"] = users
@@ -587,18 +591,28 @@ class Mutator:
                     services.append(service)
 
             elif op.op_type == "add_user":
-                users.append(
-                    {
-                        "username": str(op.params.get("username", "")),
-                        "password": str(op.params.get("password", "")),
-                        "groups": deepcopy(op.params.get("groups", [])),
-                        "hosts": deepcopy(op.params.get("hosts", [])),
-                        "email": str(op.params.get("email", "")),
-                        "full_name": str(op.params.get("full_name", "")),
-                        "department": str(op.params.get("department", "")),
-                        "role": str(op.params.get("role", "")),
-                    }
-                )
+                username = str(op.params.get("username", ""))
+                user_record = {
+                    "username": username,
+                    "password": str(op.params.get("password", "")),
+                    "groups": deepcopy(op.params.get("groups", [])),
+                    "hosts": deepcopy(op.params.get("hosts", [])),
+                    "email": str(op.params.get("email", "")),
+                    "full_name": str(op.params.get("full_name", "")),
+                    "department": str(op.params.get("department", "")),
+                    "role": str(op.params.get("role", "")),
+                }
+                users.append(user_record)
+                principal_catalog[username] = {
+                    "username": username,
+                    "kind": "user",
+                    "is_login_account": True,
+                    "hosts": deepcopy(op.params.get("hosts", [])),
+                    "department": str(op.params.get("department", "")),
+                    "role": str(op.params.get("role", "")),
+                    "email": str(op.params.get("email", "")),
+                    "full_name": str(op.params.get("full_name", "")),
+                }
 
             elif op.op_type == "add_dependency_edge":
                 dependency_edges.append(
@@ -666,34 +680,11 @@ class Mutator:
         snapshot.topology = topology
 
 
-def _build_host_catalog(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
-    catalog: dict[str, dict[str, Any]] = {}
-    for raw in manifest.get("topology", {}).get("hosts", []):
-        if not isinstance(raw, dict):
-            continue
-        name = str(raw.get("name", "")).strip()
-        if not name:
-            continue
-        catalog[name] = {
-            "zone": str(raw.get("zone", "")),
-            "services": deepcopy(raw.get("services", [])),
-            "connects_to": deepcopy(raw.get("connects_to", [])),
-        }
-    return catalog
-
-
 def _ensure_mutable_topology(
     topology: dict[str, Any],
     manifest: dict[str, Any],
 ) -> dict[str, Any]:
-    updated = dict(topology)
-    updated.setdefault("manifest_name", manifest.get("name", ""))
-    updated.setdefault("difficulty", deepcopy(manifest.get("difficulty", {})))
-    updated.setdefault("host_catalog", _build_host_catalog(manifest))
-    updated.setdefault("host_details", {})
-    updated.setdefault("dependency_edges", [])
-    updated.setdefault("trust_edges", [])
-    return updated
+    return compile_manifest_topology(manifest, topology)
 
 
 def _existing_hosts(snapshot: SnapshotSpec) -> set[str]:

src/open_range/builder/snapshot_store.py

@@ -119,6 +119,19 @@ class SnapshotStore:
             snapshot=SnapshotSpec.model_validate(raw),
         )
 
+    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),
+                )
+            )
+        return entries
+
     async def list_snapshots(self) -> list[dict[str, Any]]:
         """List all snapshots with their metadata.
 

src/open_range/lint.py

@@ -12,6 +12,7 @@ Usage::
 from __future__ import annotations
 
 import argparse
+import re
 import sys
 from pathlib import Path
 from typing import Any
@@ -134,22 +135,28 @@ def _check_business_process_flows(manifest: Manifest) -> list[str]:
     return errors
 
 
+_PRINCIPAL_RE = re.compile(r"^[A-Za-z0-9._@-]+$")
+
+
 def _check_trust_relationships(manifest: Manifest) -> list[str]:
-    """All trust relationships must reference valid usernames."""
-    user_names = {u.username for u in manifest.users}
+    """Trust principals must be well-formed identifiers.
+
+    Trust edges may reference people who are not login accounts. Those are
+    normalized into the canonical principal catalog at build time, so lint
+    should validate identifier quality rather than requiring every principal to
+    appear in ``users``.
+    """
     errors: list[str] = []
     for rel in manifest.trust_relationships:
-        if rel.source and rel.source not in user_names:
+        if rel.source and not _PRINCIPAL_RE.match(rel.source):
             errors.append(
-                f"Trust relationship source '{rel.source}' "
-                f"is not in the users list. "
-                f"Valid usernames: {sorted(user_names)}"
+                f"Trust relationship source '{rel.source}' is not a valid "
+                "principal identifier"
             )
-        if rel.target and rel.target not in user_names:
+        if rel.target and not _PRINCIPAL_RE.match(rel.target):
             errors.append(
-                f"Trust relationship target '{rel.target}' "
-                f"is not in the users list. "
-                f"Valid usernames: {sorted(user_names)}"
+                f"Trust relationship target '{rel.target}' is not a valid "
+                "principal identifier"
             )
     return errors
 
@@ -165,7 +172,7 @@ ALL_CHECKS = [
     ("NPC persona usernames", _check_npc_usernames),
     ("data inventory hosts", _check_data_inventory_hosts),
     ("business process data flows", _check_business_process_flows),
-    ("trust relationship usernames", _check_trust_relationships),
+    ("trust relationship principals", _check_trust_relationships),
 ]
 
 

src/open_range/protocols.py

@@ -66,6 +66,9 @@ class MutationPlan(BaseModel):
     predicted_complexity_delta: int = 0
     predicted_chain_delta: int = 0
     predicted_novelty: float = 0.0
+    policy_name: str = ""
+    policy_score: float = 0.0
+    score_breakdown: dict[str, float] = Field(default_factory=dict)
 
 
 class LineageMetadata(BaseModel):

src/open_range/server/runtime.py

@@ -11,6 +11,7 @@ import asyncio
 import json
 import logging
 import os
+import random
 import shlex
 import shutil
 import subprocess as sp
@@ -25,6 +26,7 @@ from typing import Any
 import yaml
 
 from open_range.builder.builder import LLMSnapshotBuilder, TemplateOnlyBuilder
+from open_range.builder.mutation_policy import PopulationMutationPolicy
 from open_range.builder.mutator import Mutator
 from open_range.builder.renderer import PAYLOAD_MANIFEST_NAME, SnapshotRenderer
 from open_range.builder.snapshot_store import SnapshotStore
@@ -41,9 +43,14 @@ from open_range.validator.build_boot import BuildBootCheck
 from open_range.validator.difficulty import DifficultyCheck
 from open_range.validator.evidence import EvidenceCheck
 from open_range.validator.exploitability import ExploitabilityCheck
+from open_range.validator.graph_consistency import GraphConsistencyCheck
+from open_range.validator.graph_evidence import GraphEvidenceSufficiencyCheck
+from open_range.validator.graph_reward_grounding import GraphRewardGroundingCheck
 from open_range.validator.isolation import IsolationCheck
+from open_range.validator.manifest_compliance import ManifestComplianceCheck
 from open_range.validator.npc_consistency import NPCConsistencyCheck
 from open_range.validator.patchability import PatchabilityCheck
+from open_range.validator.path_solvability import PathSolvabilityCheck
 from open_range.validator.realism_review import RealismReviewCheck
 from open_range.validator.reward_grounding import RewardGroundingCheck
 from open_range.validator.task_feasibility import TaskFeasibilityCheck
@@ -207,6 +214,43 @@ class CurriculumTracker:
         with self._lock:
             return list(self._history)
 
+    def snapshot_stats(self) -> dict[str, dict[str, Any]]:
+        with self._lock:
+            history = list(self._history)
+
+        now = time.time()
+        stats: dict[str, dict[str, Any]] = {}
+        for outcome in history:
+            if not outcome.snapshot_id:
+                continue
+            stat = stats.setdefault(
+                outcome.snapshot_id,
+                {
+                    "plays": 0,
+                    "completed": 0,
+                    "red_solved": 0,
+                    "blue_detected": 0,
+                    "plays_recent": 0,
+                    "last_seen_at": 0.0,
+                },
+            )
+            stat["plays"] += 1
+            if outcome.completed:
+                stat["completed"] += 1
+            if outcome.red_solved:
+                stat["red_solved"] += 1
+            if outcome.blue_detected:
+                stat["blue_detected"] += 1
+            if now - outcome.recorded_at <= 300:
+                stat["plays_recent"] += 1
+            stat["last_seen_at"] = max(float(stat["last_seen_at"]), outcome.recorded_at)
+
+        for stat in stats.values():
+            plays = max(int(stat["plays"]), 1)
+            stat["red_solve_rate"] = stat["red_solved"] / plays
+            stat["blue_detect_rate"] = stat["blue_detected"] / plays
+        return stats
+
 
 @dataclass(frozen=True, slots=True)
 class RuntimeSnapshot:
@@ -274,25 +318,38 @@ def _normalize_validator_profile(profile: str | None) -> str:
     return normalized
 
 
-def _build_validator(profile: str) -> ValidatorGate:
+def _graph_checks(manifest: dict[str, Any]) -> list[Any]:
+    return [
+        ManifestComplianceCheck(manifest),
+        GraphConsistencyCheck(),
+        PathSolvabilityCheck(),
+        GraphEvidenceSufficiencyCheck(),
+        GraphRewardGroundingCheck(),
+    ]
+
+
+def _build_validator(profile: str, manifest: dict[str, Any]) -> ValidatorGate:
     normalized = _normalize_validator_profile(profile)
     if normalized == "offline":
         return ValidatorGate(
-            [
+            _graph_checks(manifest)
+            + [
                 StructuralSnapshotCheck(),
                 TaskFeasibilityCheck(),
             ]
         )
 
     return ValidatorGate(
-        [
+        _graph_checks(manifest)
+        + [
+            StructuralSnapshotCheck(),
+            TaskFeasibilityCheck(),
             BuildBootCheck(),
             ExploitabilityCheck(),
             PatchabilityCheck(),
             EvidenceCheck(),
             RewardGroundingCheck(),
             IsolationCheck(),
-            TaskFeasibilityCheck(),
             DifficultyCheck(),
             NPCConsistencyCheck(),
             RealismReviewCheck(),
@@ -326,6 +383,7 @@ class ManagedSnapshotRuntime:
         validator_profile: str | None = None,
         pool_size: int = 3,
         selection_strategy: str = "random",
+        parent_selection_strategy: str = "policy",
         refill_enabled: bool = False,
         refill_interval_s: float = 2.0,
         generation_retries: int = 3,
@@ -334,6 +392,7 @@ class ManagedSnapshotRuntime:
         compose_runner: ComposeProjectRunner | None = None,
         live_validator: ValidatorGate | None = None,
         enable_patch_validation: bool = False,
+        mutation_policy: PopulationMutationPolicy | None = None,
     ) -> None:
         self.manifest_path = (
             Path(manifest_path).resolve()
@@ -344,15 +403,17 @@ class ManagedSnapshotRuntime:
         self.store_dir = _resolve_store_dir(store_dir)
         self.store = SnapshotStore(str(self.store_dir))
         self.builder = builder or _default_builder()
-        self.mutator = Mutator(self.builder)
+        self.mutation_policy = mutation_policy or PopulationMutationPolicy()
+        self.mutator = Mutator(self.builder, policy=self.mutation_policy)
         self.validator_profile = _normalize_validator_profile(
             validator_profile or os.getenv("OPENRANGE_RUNTIME_VALIDATOR_PROFILE", "offline")
         )
-        self.validator = validator or _build_validator(self.validator_profile)
+        self.validator = validator or _build_validator(self.validator_profile, self.manifest)
         self.renderer = SnapshotRenderer()
         self.curriculum = CurriculumTracker()
         self.pool_size = max(1, pool_size)
         self.selection_strategy = selection_strategy
+        self.parent_selection_strategy = parent_selection_strategy
         self.refill_enabled = refill_enabled
         self.refill_interval_s = max(0.25, refill_interval_s)
         self.generation_retries = max(1, generation_retries)
@@ -381,6 +442,7 @@ class ManagedSnapshotRuntime:
             validator_profile=os.getenv("OPENRANGE_RUNTIME_VALIDATOR_PROFILE", "offline"),
             pool_size=_env_int("OPENRANGE_SNAPSHOT_POOL_SIZE", 3),
             selection_strategy=os.getenv("OPENRANGE_SNAPSHOT_SELECTION", "random"),
+            parent_selection_strategy=os.getenv("OPENRANGE_PARENT_SELECTION", "policy"),
             refill_enabled=_env_flag("OPENRANGE_ENABLE_MANAGED_REFILL", default=False),
             refill_interval_s=float(os.getenv("OPENRANGE_REFILL_INTERVAL_S", "2.0")),
             generation_retries=_env_int("OPENRANGE_GENERATION_RETRIES", 3),
@@ -541,6 +603,7 @@ class ManagedSnapshotRuntime:
             "store_dir": str(self.store_dir),
             "pool_size": self.pool_size,
             "selection_strategy": self.selection_strategy,
+            "parent_selection_strategy": self.parent_selection_strategy,
             "validator_profile": self.validator_profile,
             "refill_enabled": self.refill_enabled,
             "live_admission_enabled": self.live_admission_enabled,
@@ -623,7 +686,7 @@ class ManagedSnapshotRuntime:
 
         for attempt in range(1, self.generation_retries + 1):
             context = self._build_context()
-            parent_entry = self._select_parent_entry()
+            parent_entry = self._select_parent_entry(context)
             snapshot = _run_coro_sync(
                 self.mutator.mutate(
                     self.manifest,
@@ -930,10 +993,29 @@ class ManagedSnapshotRuntime:
         prefix = "snap_" + "_".join(vuln_types[:3]) if vuln_types else "snap_generated"
         return f"{prefix}_{int(time.time() * 1000)}"
 
-    def _select_parent_entry(self):
+    def _select_parent_entry(self, context: BuildContext):
         if self.snapshot_count() == 0:
             return None
-        return _run_coro_sync(self.store.select_entry(strategy=self.selection_strategy))
+        if self.parent_selection_strategy in {"latest", "random"}:
+            return _run_coro_sync(self.store.select_entry(strategy=self.parent_selection_strategy))
+        entries = _run_coro_sync(self.store.list_entries())
+        if not entries:
+            return None
+        rng = random.Random(context.seed if context.seed is not None else self._generation_counter)
+        selected, score = self.mutation_policy.select_parent(
+            entries,
+            context=context,
+            snapshot_stats=self.curriculum.snapshot_stats(),
+            rng=rng,
+        )
+        logger.info(
+            "ManagedSnapshotRuntime selected parent %s via %s (score=%.3f components=%s)",
+            selected.snapshot_id,
+            self.mutation_policy.name,
+            score.total,
+            score.components,
+        )
+        return selected
 
     def _snapshot_dir(self, snapshot_id: str) -> Path:
         return self.store_dir / snapshot_id

src/open_range/validator/graph_consistency.py

@@ -18,8 +18,8 @@ class GraphConsistencyCheck:
                 issues.append(f"dependency edge '{source}->{target}' references unknown host")
 
         for source, target, _edge_type in compiled.trust_edges:
-            if source not in compiled.users or target not in compiled.users:
-                issues.append(f"trust edge '{source}->{target}' references unknown user")
+            if source not in compiled.principals or target not in compiled.principals:
+                issues.append(f"trust edge '{source}->{target}' references unknown principal")
 
         lineage = snapshot.lineage
         if lineage.generation_depth == 0 and lineage.parent_snapshot_id:
@@ -50,13 +50,13 @@ class GraphConsistencyCheck:
                 if op.op_type == "add_trust_edge":
                     source = op.target_selector.get("source", "")
                     target = op.target_selector.get("target", "")
-                    if source and source not in compiled.users:
+                    if source and source not in compiled.principals:
                         issues.append(
-                            f"mutation '{op.mutation_id}' source user '{source}' missing"
+                            f"mutation '{op.mutation_id}' source principal '{source}' missing"
                         )
-                    if target and target not in compiled.users:
+                    if target and target not in compiled.principals:
                         issues.append(
-                            f"mutation '{op.mutation_id}' target user '{target}' missing"
+                            f"mutation '{op.mutation_id}' target principal '{target}' missing"
                         )
 
         passed = len(issues) == 0
@@ -66,6 +66,7 @@ class GraphConsistencyCheck:
             details={
                 "hosts": len(compiled.hosts),
                 "users": len(compiled.users),
+                "principals": len(compiled.principals),
                 "dependency_edges": len(compiled.dependency_edges),
                 "trust_edges": len(compiled.trust_edges),
             },

src/open_range/validator/graph_evidence.py

@@ -0,0 +1,57 @@
+"""Graph-native evidence sufficiency checks."""
+
+from __future__ import annotations
+
+from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
+from open_range.validator.graphs import compile_snapshot_graphs
+
+
+class GraphEvidenceSufficiencyCheck:
+    """Verify that the compiled world exposes enough evidence for key facts."""
+
+    async def check(self, snapshot: SnapshotSpec, containers: ContainerSet) -> CheckResult:
+        compiled = compile_snapshot_graphs(snapshot)
+        evidence_hosts = {
+            _location_host(location)
+            for location in compiled.evidence_locations
+            if _location_host(location)
+        }
+        issues: list[str] = []
+
+        if not compiled.evidence_locations:
+            return CheckResult(
+                name="graph_evidence_sufficiency",
+                passed=False,
+                error="snapshot has no evidence locations",
+            )
+
+        for vuln in snapshot.truth_graph.vulns:
+            supporting_hosts = {vuln.host, "siem"}
+            if not evidence_hosts.intersection(supporting_hosts):
+                issues.append(
+                    f"vuln '{vuln.id}' on host '{vuln.host}' has no supporting evidence host"
+                )
+
+        for flag in snapshot.flags:
+            supporting_hosts = {flag.host, "siem"}
+            if not evidence_hosts.intersection(supporting_hosts):
+                issues.append(
+                    f"flag '{flag.id}' on host '{flag.host}' has no supporting evidence host"
+                )
+
+        passed = len(issues) == 0
+        return CheckResult(
+            name="graph_evidence_sufficiency",
+            passed=passed,
+            details={
+                "evidence_hosts": sorted(evidence_hosts),
+                "issues": issues,
+            },
+            error="" if passed else "; ".join(issues),
+        )
+
+
+def _location_host(location: str) -> str:
+    if ":" not in location:
+        return "siem"
+    return location.split(":", 1)[0].strip()

src/open_range/validator/graph_reward_grounding.py

@@ -0,0 +1,47 @@
+"""Graph-native reward grounding checks."""
+
+from __future__ import annotations
+
+from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
+from open_range.validator.graphs import compile_snapshot_graphs
+from open_range.validator.path_solvability import build_host_adjacency, has_host_path
+
+
+class GraphRewardGroundingCheck:
+    """Verify rewards are grounded by graph facts before live checks run."""
+
+    async def check(self, snapshot: SnapshotSpec, containers: ContainerSet) -> CheckResult:
+        compiled = compile_snapshot_graphs(snapshot)
+        issues: list[str] = []
+
+        if not snapshot.flags:
+            return CheckResult(
+                name="graph_reward_grounding",
+                passed=False,
+                error="snapshot has no flags to ground",
+            )
+
+        adjacency = build_host_adjacency(snapshot, compiled)
+        vuln_hosts = {v.host for v in snapshot.truth_graph.vulns if v.host}
+        for flag in snapshot.flags:
+            if flag.host not in compiled.hosts:
+                issues.append(f"flag '{flag.id}' references unknown host '{flag.host}'")
+                continue
+
+            if flag.host in vuln_hosts:
+                continue
+
+            if vuln_hosts and not any(
+                has_host_path(source, flag.host, adjacency) for source in vuln_hosts
+            ):
+                issues.append(
+                    f"flag '{flag.id}' on '{flag.host}' is not reachable from any vuln host"
+                )
+
+        passed = len(issues) == 0
+        return CheckResult(
+            name="graph_reward_grounding",
+            passed=passed,
+            details={"issues": issues},
+            error="" if passed else "; ".join(issues),
+        )

src/open_range/validator/graphs.py

@@ -18,6 +18,8 @@ class CompiledGraphs:
 
     hosts: frozenset[str]
     users: frozenset[str]
+    principals: frozenset[str]
+    zones_by_host: dict[str, str]
     services_by_host: dict[str, frozenset[str]]
     dependency_edges: frozenset[tuple[str, str]]
     trust_edges: frozenset[tuple[str, str, str]]
@@ -31,6 +33,8 @@ def compile_snapshot_graphs(snapshot: SnapshotSpec) -> CompiledGraphs:
     topology = snapshot.topology or {}
     hosts = _compile_hosts(topology)
     users = _compile_users(topology)
+    principals = _compile_principals(topology, users)
+    zones_by_host = _compile_zones(topology, hosts)
     services_by_host = _compile_services(topology, hosts)
     dependency_edges = _compile_dependency_edges(topology)
     trust_edges = _compile_trust_edges(topology)
@@ -40,6 +44,8 @@ def compile_snapshot_graphs(snapshot: SnapshotSpec) -> CompiledGraphs:
     return CompiledGraphs(
         hosts=hosts,
         users=users,
+        principals=principals,
+        zones_by_host=zones_by_host,
         services_by_host=services_by_host,
         dependency_edges=dependency_edges,
         trust_edges=trust_edges,
@@ -80,6 +86,7 @@ def _compile_services(
     hosts: frozenset[str],
 ) -> dict[str, frozenset[str]]:
     host_details = topology.get("host_details", {})
+    host_catalog = topology.get("host_catalog", {})
     compiled: dict[str, frozenset[str]] = {}
     for host in hosts:
         detail = {}
@@ -87,6 +94,10 @@ def _compile_services(
             raw_detail = host_details.get(host, {})
             if isinstance(raw_detail, dict):
                 detail = raw_detail
+        if not detail and isinstance(host_catalog, dict):
+            raw_catalog = host_catalog.get(host, {})
+            if isinstance(raw_catalog, dict):
+                detail = raw_catalog
         services = detail.get("services", [])
         if not isinstance(services, list):
             services = []
@@ -104,6 +115,21 @@ def _compile_dependency_edges(topology: dict[str, object]) -> frozenset[tuple[st
         target = str(raw.get("target", "")).strip()
         if source and target:
             edges.add((source, target))
+    if edges:
+        return frozenset(edges)
+
+    host_details = topology.get("host_details", {})
+    if isinstance(host_details, dict):
+        for source, raw_detail in host_details.items():
+            if not isinstance(raw_detail, dict):
+                continue
+            raw_targets = raw_detail.get("connects_to", [])
+            if not isinstance(raw_targets, list):
+                continue
+            for raw_target in raw_targets:
+                target = str(raw_target).strip()
+                if source and target:
+                    edges.add((str(source).strip(), target))
     return frozenset(edges)
 
 
@@ -119,3 +145,45 @@ def _compile_trust_edges(topology: dict[str, object]) -> frozenset[tuple[str, st
         if source and target:
             edges.add((source, target, edge_type))
     return frozenset(edges)
+
+
+def _compile_principals(
+    topology: dict[str, object],
+    users: frozenset[str],
+) -> frozenset[str]:
+    principals = set(users)
+    raw_catalog = topology.get("principal_catalog", {})
+    if isinstance(raw_catalog, dict):
+        for raw_name in raw_catalog:
+            name = str(raw_name).strip()
+            if name:
+                principals.add(name)
+    return frozenset(principals)
+
+
+def _compile_zones(
+    topology: dict[str, object],
+    hosts: frozenset[str],
+) -> dict[str, str]:
+    zones_by_host: dict[str, str] = {}
+    raw_zones = topology.get("zones", {})
+    if isinstance(raw_zones, dict):
+        for raw_zone, raw_hosts in raw_zones.items():
+            zone = str(raw_zone).strip()
+            if not zone or not isinstance(raw_hosts, list):
+                continue
+            for raw_host in raw_hosts:
+                host = str(raw_host).strip()
+                if host:
+                    zones_by_host[host] = zone
+
+    host_details = topology.get("host_details", {})
+    if isinstance(host_details, dict):
+        for raw_host, raw_detail in host_details.items():
+            host = str(raw_host).strip()
+            if not host or host not in hosts or not isinstance(raw_detail, dict):
+                continue
+            zone = str(raw_detail.get("zone", "")).strip()
+            if zone and host not in zones_by_host:
+                zones_by_host[host] = zone
+    return zones_by_host

src/open_range/validator/manifest_compliance.py

@@ -32,6 +32,7 @@ class ManifestComplianceCheck:
         manifest_hosts = _manifest_hosts(self.manifest)
         allowed_bug_families = set(str(v) for v in self.manifest.get("bug_families", []))
         allowed_users = set(_manifest_users(self.manifest))
+        allowed_principals = set(_manifest_principals(self.manifest))
         allowed_services = _manifest_services(self.manifest)
         allowed_dependency_edges = _manifest_dependency_edges(self.manifest)
         allowed_trust_edges = _manifest_trust_edges(self.manifest)
@@ -55,6 +56,20 @@ class ManifestComplianceCheck:
                     f"host '{host}' has services outside manifest family: {sorted(illegal)}"
                 )
 
+        illegal_dependency_edges = sorted(
+            edge for edge in compiled.dependency_edges if edge not in allowed_dependency_edges
+        )
+        if illegal_dependency_edges:
+            issues.append(
+                f"dependency edges outside manifest family: {illegal_dependency_edges}"
+            )
+
+        illegal_trust_edges = sorted(
+            edge for edge in compiled.trust_edges if edge not in allowed_trust_edges
+        )
+        if illegal_trust_edges:
+            issues.append(f"trust edges outside manifest family: {illegal_trust_edges}")
+
         for vuln in snapshot.truth_graph.vulns:
             if vuln.type and allowed_bug_families and vuln.type not in allowed_bug_families:
                 issues.append(f"vuln '{vuln.id}' uses disallowed family '{vuln.type}'")
@@ -93,6 +108,14 @@ class ManifestComplianceCheck:
                     source = op.target_selector.get("source", "")
                     target = op.target_selector.get("target", "")
                     edge_type = str(op.params.get("type", "")).strip()
+                    if source and source not in allowed_principals:
+                        issues.append(
+                            f"add_trust_edge introduces unknown principal '{source}'"
+                        )
+                    if target and target not in allowed_principals:
+                        issues.append(
+                            f"add_trust_edge introduces unknown principal '{target}'"
+                        )
                     if (source, target, edge_type) not in allowed_trust_edges:
                         issues.append(
                             f"add_trust_edge introduces illegal trust edge "
@@ -139,6 +162,20 @@ def _manifest_users(manifest: dict[str, Any]) -> set[str]:
     return users
 
 
+def _manifest_principals(manifest: dict[str, Any]) -> set[str]:
+    principals = set(_manifest_users(manifest))
+    for raw in manifest.get("trust_relationships", []):
+        if not isinstance(raw, dict):
+            continue
+        source = str(raw.get("source") or raw.get("from") or "").strip()
+        target = str(raw.get("target") or raw.get("to") or "").strip()
+        if source:
+            principals.add(source)
+        if target:
+            principals.add(target)
+    return principals
+
+
 def _manifest_services(manifest: dict[str, Any]) -> dict[str, frozenset[str]]:
     services: dict[str, frozenset[str]] = {}
     for raw in manifest.get("topology", {}).get("hosts", []):

src/open_range/validator/path_solvability.py

@@ -0,0 +1,164 @@
+"""Graph-native path solvability checks."""
+
+from __future__ import annotations
+
+from collections import defaultdict, deque
+
+from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
+from open_range.validator.graphs import CompiledGraphs, compile_snapshot_graphs
+
+
+class PathSolvabilityCheck:
+    """Verify that vuln and flag hosts are reachable in the compiled host graph."""
+
+    async def check(self, snapshot: SnapshotSpec, containers: ContainerSet) -> CheckResult:
+        compiled = compile_snapshot_graphs(snapshot)
+        issues: list[str] = []
+
+        if not compiled.hosts:
+            return CheckResult(
+                name="path_solvability",
+                passed=False,
+                error="snapshot has no compiled hosts",
+            )
+
+        start_hosts = _start_hosts(compiled)
+        vuln_hosts = {v.host for v in snapshot.truth_graph.vulns if v.host}
+        flag_hosts = {flag.host for flag in snapshot.flags if flag.host}
+        target_hosts = sorted(vuln_hosts.union(flag_hosts))
+        if not target_hosts:
+            return CheckResult(
+                name="path_solvability",
+                passed=False,
+                error="snapshot has no vuln or flag hosts to solve toward",
+            )
+
+        adjacency = build_host_adjacency(snapshot, compiled)
+        unreachable = [
+            host
+            for host in target_hosts
+            if not _reachable_from_any(host, start_hosts, adjacency)
+        ]
+        if unreachable:
+            issues.append(f"unreachable target hosts from start set {sorted(start_hosts)}: {unreachable}")
+
+        for flag_host in sorted(flag_hosts):
+            if not (
+                flag_host in vuln_hosts
+                or _reachable_from_any(flag_host, vuln_hosts or start_hosts, adjacency)
+            ):
+                issues.append(
+                    f"flag host '{flag_host}' is not grounded by any vuln host or start host"
+                )
+
+        passed = len(issues) == 0
+        return CheckResult(
+            name="path_solvability",
+            passed=passed,
+            details={
+                "start_hosts": sorted(start_hosts),
+                "target_hosts": target_hosts,
+                "issues": issues,
+            },
+            error="" if passed else "; ".join(issues),
+        )
+
+
+def _start_hosts(compiled: CompiledGraphs) -> set[str]:
+    starts = {
+        host
+        for host in compiled.hosts
+        if host in {"attacker", "internet"}
+        or compiled.zones_by_host.get(host) == "external"
+    }
+    if starts:
+        return starts
+    if compiled.hosts:
+        return {sorted(compiled.hosts)[0]}
+    return set()
+
+
+def build_host_adjacency(
+    snapshot: SnapshotSpec,
+    compiled: CompiledGraphs,
+) -> dict[str, set[str]]:
+    adjacency: dict[str, set[str]] = defaultdict(set)
+    for source, target in compiled.dependency_edges:
+        adjacency[source].add(target)
+
+    principal_hosts = _principal_hosts(snapshot)
+    for source_principal, target_principal, _edge_type in compiled.trust_edges:
+        source_hosts = principal_hosts.get(source_principal, set())
+        target_hosts = principal_hosts.get(target_principal, set())
+        for source_host in source_hosts:
+            for target_host in target_hosts:
+                if source_host and target_host:
+                    adjacency[source_host].add(target_host)
+    return adjacency
+
+
+def has_host_path(
+    start: str,
+    target: str,
+    adjacency: dict[str, set[str]],
+) -> bool:
+    return _has_path(start, target, adjacency)
+
+
+def _principal_hosts(snapshot: SnapshotSpec) -> dict[str, set[str]]:
+    topology = snapshot.topology or {}
+    mapping: dict[str, set[str]] = defaultdict(set)
+
+    raw_users = topology.get("users", [])
+    if isinstance(raw_users, list):
+        for raw in raw_users:
+            if not isinstance(raw, dict):
+                continue
+            username = str(raw.get("username", "")).strip()
+            if not username:
+                continue
+            for raw_host in raw.get("hosts", []):
+                host = str(raw_host).strip()
+                if host:
+                    mapping[username].add(host)
+
+    raw_catalog = topology.get("principal_catalog", {})
+    if isinstance(raw_catalog, dict):
+        for raw_name, raw_principal in raw_catalog.items():
+            name = str(raw_name).strip()
+            if not name or not isinstance(raw_principal, dict):
+                continue
+            for raw_host in raw_principal.get("hosts", []):
+                host = str(raw_host).strip()
+                if host:
+                    mapping[name].add(host)
+
+    return mapping
+
+
+def _reachable_from_any(
+    target: str,
+    starts: set[str],
+    adjacency: dict[str, set[str]],
+) -> bool:
+    for start in starts:
+        if start == target:
+            return True
+        if _has_path(start, target, adjacency):
+            return True
+    return False
+
+
+def _has_path(start: str, target: str, adjacency: dict[str, set[str]]) -> bool:
+    queue: deque[str] = deque([start])
+    seen = {start}
+    while queue:
+        current = queue.popleft()
+        for neighbor in adjacency.get(current, set()):
+            if neighbor == target:
+                return True
+            if neighbor in seen:
+                continue
+            seen.add(neighbor)
+            queue.append(neighbor)
+    return False

tests/test_builder.py

@@ -174,6 +174,25 @@ async def test_mutator_builds_child_snapshot_with_lineage(tier1_manifest):
     assert child.lineage.mutation_summary
 
 
+@pytest.mark.asyncio
+async def test_mutator_compiles_root_snapshot_from_manifest_graph(tier1_manifest):
+    from open_range.builder.builder import TemplateOnlyBuilder
+    from open_range.builder.mutator import Mutator
+
+    root = await Mutator(TemplateOnlyBuilder()).mutate(
+        tier1_manifest,
+        context=BuildContext(seed=1, tier=1),
+    )
+    topology = root.topology
+    assert topology["host_details"]["web"]["services"]
+    assert topology["dependency_edges"]
+    assert topology["trust_edges"]
+    assert "principal_catalog" in topology
+    assert "schen" in topology["principal_catalog"]
+    assert "schen" not in {user["username"] for user in topology["users"]}
+    assert topology["manifest_normalization"]["trust_only_principals"]
+
+
 @pytest.mark.asyncio
 async def test_mutator_rebuilds_child_files_from_mutated_snapshot(tier1_manifest):
     from open_range.builder.builder import TemplateOnlyBuilder

tests/test_lint.py

@@ -106,9 +106,10 @@ class TestValidManifest:
             assert errors == [], f"Check '{check_name}' failed: {errors}"
 
     def test_tier1_manifest_loads(self):
-        """Tier 1 manifest should at least load without schema error."""
+        """Tier 1 manifest should load and pass lint checks."""
         result = lint_file(ROOT / "manifests" / "tier1_basic.yaml")
         assert result["schema_error"] is None, result["schema_error"]
+        assert result["valid"] is True, result["checks"]
 
 
 # ---------------------------------------------------------------------------
@@ -177,35 +178,35 @@ class TestInvalidUserRefs:
         assert len(errors) == 1
         assert "ghost_user" in errors[0]
 
-    def test_trust_relationship_invalid_source(self):
+    def test_trust_relationship_invalid_source_identifier(self):
         data = _minimal_manifest()
         data["trust_relationships"] = [
             {
                 "type": "delegates_access",
-                "from": "nobody",
+                "from": "bad actor!",
                 "to": "admin",
             },
         ]
         manifest = Manifest(**data)
         results = lint_manifest(manifest)
-        errors = results["trust relationship usernames"]
+        errors = results["trust relationship principals"]
         assert len(errors) == 1
-        assert "nobody" in errors[0]
+        assert "bad actor!" in errors[0]
 
-    def test_trust_relationship_invalid_target(self):
+    def test_trust_relationship_invalid_target_identifier(self):
         data = _minimal_manifest()
         data["trust_relationships"] = [
             {
                 "type": "delegates_access",
                 "from": "admin",
-                "to": "phantom",
+                "to": "phantom user",
             },
         ]
         manifest = Manifest(**data)
         results = lint_manifest(manifest)
-        errors = results["trust relationship usernames"]
+        errors = results["trust relationship principals"]
         assert len(errors) == 1
-        assert "phantom" in errors[0]
+        assert "phantom user" in errors[0]
 
 
 # ---------------------------------------------------------------------------

tests/test_runtime.py

@@ -21,6 +21,11 @@ class TestManagedSnapshotRuntime:
         )
         names = [type(check).__name__ for check in runtime.validator.checks]
         assert names == [
+            "ManifestComplianceCheck",
+            "GraphConsistencyCheck",
+            "PathSolvabilityCheck",
+            "GraphEvidenceSufficiencyCheck",
+            "GraphRewardGroundingCheck",
             "StructuralSnapshotCheck",
             "TaskFeasibilityCheck",
         ]
@@ -33,6 +38,13 @@ class TestManagedSnapshotRuntime:
             refill_enabled=False,
         )
         names = [type(check).__name__ for check in runtime.validator.checks]
+        assert names[:5] == [
+            "ManifestComplianceCheck",
+            "GraphConsistencyCheck",
+            "PathSolvabilityCheck",
+            "GraphEvidenceSufficiencyCheck",
+            "GraphRewardGroundingCheck",
+        ]
         assert "BuildBootCheck" in names
         assert "ExploitabilityCheck" in names
         assert "PatchabilityCheck" in names
@@ -117,6 +129,7 @@ class TestManagedSnapshotRuntime:
             store_dir=tmp_path / "snapshots",
             pool_size=2,
             selection_strategy="latest",
+            parent_selection_strategy="policy",
             refill_enabled=False,
         )
 
@@ -131,6 +144,17 @@ class TestManagedSnapshotRuntime:
         finally:
             runtime.stop()
 
+    def test_status_reports_parent_selection_strategy(self, tier1_manifest, tmp_path):
+        runtime = ManagedSnapshotRuntime(
+            manifest=tier1_manifest,
+            store_dir=tmp_path / "snapshots",
+            pool_size=1,
+            parent_selection_strategy="policy",
+            refill_enabled=False,
+        )
+        status = runtime.status()
+        assert status["parent_selection_strategy"] == "policy"
+
     def test_acquire_snapshot_exposes_lineage_metadata(self, tier1_manifest, tmp_path):
         runtime = ManagedSnapshotRuntime(
             manifest=tier1_manifest,

tests/test_validator.py

@@ -73,6 +73,134 @@ async def test_graph_consistency_rejects_missing_parent_lineage(sample_snapshot_
     assert "missing parent_snapshot_id" in result.error
 
 
+@pytest.mark.asyncio
+async def test_path_solvability_passes_for_reachable_flag_host(mock_containers):
+    from open_range.protocols import EvidenceItem, TruthGraph, Vulnerability
+    from open_range.validator.path_solvability import PathSolvabilityCheck
+
+    spec = SnapshotSpec(
+        topology={
+            "hosts": ["attacker", "web", "db"],
+            "zones": {"external": ["attacker"], "dmz": ["web"], "internal": ["db"]},
+            "dependency_edges": [
+                {"source": "attacker", "target": "web"},
+                {"source": "web", "target": "db"},
+            ],
+            "host_details": {
+                "attacker": {"services": ["nmap"]},
+                "web": {"services": ["nginx"]},
+                "db": {"services": ["mysql"]},
+            },
+        },
+        truth_graph=TruthGraph(
+            vulns=[Vulnerability(id="v1", type="sqli", host="web", service="nginx")],
+        ),
+        flags=[FlagSpec(id="f1", value="FLAG{ok}", path="/var/flags/flag1.txt", host="db")],
+        evidence_spec=[EvidenceItem(type="log_entry", location="siem:/var/log/siem/all.log")],
+        golden_path=[GoldenPathStep(step=1, command="nmap web", expect_in_stdout="80/tcp")],
+        task=TaskSpec(red_briefing="go", blue_briefing="watch"),
+    )
+
+    result = await PathSolvabilityCheck().check(spec, mock_containers)
+    assert result.passed is True
+
+
+@pytest.mark.asyncio
+async def test_graph_evidence_sufficiency_fails_without_supporting_hosts(mock_containers):
+    from open_range.protocols import TruthGraph, Vulnerability
+    from open_range.validator.graph_evidence import GraphEvidenceSufficiencyCheck
+
+    spec = SnapshotSpec(
+        topology={
+            "hosts": ["attacker", "web", "db"],
+            "zones": {"external": ["attacker"], "dmz": ["web"], "internal": ["db"]},
+            "dependency_edges": [{"source": "attacker", "target": "web"}],
+            "host_details": {
+                "attacker": {"services": ["nmap"]},
+                "web": {"services": ["nginx"]},
+                "db": {"services": ["mysql"]},
+            },
+        },
+        truth_graph=TruthGraph(
+            vulns=[Vulnerability(id="v1", type="sqli", host="db", service="mysql")],
+        ),
+        flags=[FlagSpec(id="f1", value="FLAG{db}", path="/var/flags/flag1.txt", host="db")],
+        evidence_spec=[EvidenceItem(type="log_entry", location="web:/var/log/access.log")],
+        golden_path=[GoldenPathStep(step=1, command="scan", expect_in_stdout="ok")],
+        task=TaskSpec(red_briefing="go", blue_briefing="watch"),
+    )
+
+    result = await GraphEvidenceSufficiencyCheck().check(spec, mock_containers)
+    assert result.passed is False
+    assert "no supporting evidence host" in result.error
+
+
+@pytest.mark.asyncio
+async def test_graph_reward_grounding_fails_when_flag_host_unreachable(mock_containers):
+    from open_range.protocols import TruthGraph, Vulnerability
+    from open_range.validator.graph_reward_grounding import GraphRewardGroundingCheck
+
+    spec = SnapshotSpec(
+        topology={
+            "hosts": ["attacker", "web", "db"],
+            "zones": {"external": ["attacker"], "dmz": ["web"], "internal": ["db"]},
+            "dependency_edges": [{"source": "attacker", "target": "web"}],
+            "host_details": {
+                "attacker": {"services": ["nmap"]},
+                "web": {"services": ["nginx"]},
+                "db": {"services": ["mysql"]},
+            },
+        },
+        truth_graph=TruthGraph(
+            vulns=[Vulnerability(id="v1", type="sqli", host="web", service="nginx")],
+        ),
+        flags=[FlagSpec(id="f1", value="FLAG{db}", path="/var/flags/flag1.txt", host="db")],
+        evidence_spec=[EvidenceItem(type="log_entry", location="siem:/var/log/siem/all.log")],
+        golden_path=[GoldenPathStep(step=1, command="scan", expect_in_stdout="ok")],
+        task=TaskSpec(red_briefing="go", blue_briefing="watch"),
+    )
+
+    result = await GraphRewardGroundingCheck().check(spec, mock_containers)
+    assert result.passed is False
+    assert "not reachable from any vuln host" in result.error
+
+
+@pytest.mark.asyncio
+async def test_graph_checks_allow_trust_based_host_pivots(mock_containers):
+    from open_range.validator.graph_reward_grounding import GraphRewardGroundingCheck
+    from open_range.validator.path_solvability import PathSolvabilityCheck
+
+    spec = SnapshotSpec(
+        topology={
+            "hosts": ["attacker", "web", "db"],
+            "zones": {"external": ["attacker"], "dmz": ["web"], "internal": ["db"]},
+            "dependency_edges": [{"source": "attacker", "target": "web"}],
+            "trust_edges": [{"source": "websvc", "target": "dbsvc", "type": "credential_reuse"}],
+            "host_details": {
+                "attacker": {"services": ["nmap"]},
+                "web": {"services": ["nginx"]},
+                "db": {"services": ["mysql"]},
+            },
+            "principal_catalog": {
+                "websvc": {"username": "websvc", "hosts": ["web"], "is_login_account": False},
+                "dbsvc": {"username": "dbsvc", "hosts": ["db"], "is_login_account": False},
+            },
+        },
+        truth_graph=TruthGraph(
+            vulns=[Vulnerability(id="v1", type="credential_reuse", host="web", service="nginx")],
+        ),
+        flags=[FlagSpec(id="f1", value="FLAG{db}", path="/var/flags/flag1.txt", host="db")],
+        evidence_spec=[EvidenceItem(type="log_entry", location="db:/var/log/mysql.log")],
+        golden_path=[GoldenPathStep(step=1, command="scan", expect_in_stdout="ok")],
+        task=TaskSpec(red_briefing="go", blue_briefing="watch"),
+    )
+
+    path_result = await PathSolvabilityCheck().check(spec, mock_containers)
+    reward_result = await GraphRewardGroundingCheck().check(spec, mock_containers)
+    assert path_result.passed is True
+    assert reward_result.passed is True
+
+
 # ---------------------------------------------------------------------------
 # Check 1: BuildBoot
 # ---------------------------------------------------------------------------