Add snapshot lineage and parent-based mutation runtime (#49)

README.md

@@ -12,15 +12,17 @@ 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, vulnerability classes, difficulty. A shared **ManagedSnapshotRuntime** inside the shipped OpenEnv server process owns the snapshot pool. It loads admitted snapshots from disk or preloads a deterministic pool from the manifest, renders each admitted snapshot into concrete Docker artifacts under `snapshots/<id>/artifacts`, then optionally refills that pool in the background. `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 root snapshot from the manifest, then derives child snapshots by applying explicit typed mutations to admitted parents. Each candidate child is validated before render/materialization and then stored under `snapshots/<id>/`. `reset()` selects one frozen admitted snapshot. `step()` runs commands inside it.
 
 ```mermaid
 flowchart LR
-    M[Manifest<br/>topology, services,<br/>bug families, difficulty] --> R[ManagedSnapshotRuntime<br/>shared inside server process]
-    R --> B[Builder / mutator<br/>deterministic by default,<br/>LiteLLM optional]
-    B --> V{Validator}
-    V -->|fail| B
-    V -->|pass| S[Frozen admitted snapshots]
+    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]
+    U --> V{Validator<br/>manifest + graph +<br/>runtime checks}
+    V -->|fail| U
+    V -->|pass| S[Admitted snapshot population]
     S --> E["reset() → step() → obs + reward"]
 
     style V fill:#ffd93d,color:#333
@@ -64,15 +66,15 @@ uv run pytest tests/ -v --tb=short
 
 ## Core Components
 
-**Manifest** — YAML defining the legal world: hosts, zones, services, users, NPCs, data assets, credential policies, monitoring coverage, trust relationships, and which vulnerability classes the Builder may plant. Three example manifests ship (healthcare, fintech, SaaS) at tiers 1-3.
+**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`, builder/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, 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** — Takes a manifest + curriculum context, outputs a `SnapshotSpec`: topology graph, truth graph (planted vulns + exploit chain), evidence graph (what Blue can find), flags, golden path, NPC traffic, and task briefings. Three implementations: `LLMSnapshotBuilder` (production, via litellm), `TemplateOnlyBuilder` (deterministic shipped default), `FileBuilder` (load from disk).
+**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). 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 uses structural checks that operate on the compiled `SnapshotSpec` without requiring live model calls; richer container-backed checks remain available for private/local generation workflows.
+**Validator** — Admission gate for candidate snapshots. The shipped runtime now enforces manifest compliance and graph consistency before structural/task checks. These mechanical checks operate on the compiled `SnapshotSpec` without requiring live model calls; richer container-backed checks remain available for private/local generation workflows.
 
 **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/mutator.py

@@ -8,12 +8,44 @@ context from failed validations.
 from __future__ import annotations
 
 import logging
+import random
+from copy import deepcopy
 from typing import Any
 
-from open_range.protocols import BuildContext, SnapshotBuilder, SnapshotSpec
+from open_range.protocols import (
+    BuildContext,
+    EvidenceItem,
+    ExploitStep,
+    LineageMetadata,
+    MutationOp,
+    MutationPlan,
+    SnapshotBuilder,
+    SnapshotSpec,
+    Vulnerability,
+)
 
 logger = logging.getLogger(__name__)
 
+_SUPPORTED_MUTATION_OPS = {
+    "add_service",
+    "add_user",
+    "add_dependency_edge",
+    "add_trust_edge",
+    "seed_vuln",
+    "add_benign_noise",
+}
+
+_INJECTION_POINTS = {
+    "sqli": "/legacy/search.php?q=",
+    "idor": "/api/users/{id}",
+    "path_traversal": "/download?file=",
+    "command_injection": "/admin/diagnostics?host=",
+    "ssrf": "/fetch?url=",
+    "weak_creds": "ssh svc_app@host",
+    "broken_auth": "/admin/login",
+    "xss": "/search?q=",
+}
+
 
 class Mutator:
     """Orchestrate vuln mutation across resets.
@@ -38,16 +70,21 @@ class Mutator:
         manifest: dict,
         context: BuildContext | None = None,
         error: dict[str, Any] | None = None,
+        parent_snapshot: SnapshotSpec | None = None,
+        parent_snapshot_id: str | None = None,
     ) -> SnapshotSpec:
-        """Generate a mutated snapshot, avoiding recent vuln classes.
+        """Generate a root or child snapshot, avoiding recent vuln classes.
 
         Args:
             manifest: Parsed manifest dict.
             context: Optional base context (curriculum stats, etc.).
             error: Error feedback from a failed validation attempt.
+            parent_snapshot: Admitted parent snapshot to mutate forward.
+            parent_snapshot_id: Persisted ID for *parent_snapshot*.
 
         Returns:
-            A new SnapshotSpec with different vulns from the previous episode.
+            A new SnapshotSpec. Root snapshots are compiled from the manifest;
+            child snapshots are mutated from the parent.
         """
         if context is None:
             context = BuildContext()
@@ -64,7 +101,16 @@ class Mutator:
             except (AttributeError, ValueError):
                 pass  # protocol version without error field
 
-        snapshot = await self.builder.build(manifest, context)
+        if parent_snapshot is None:
+            snapshot = await self.builder.build(manifest, context)
+            snapshot = self._hydrate_root_snapshot(snapshot, manifest)
+        else:
+            snapshot = self._mutate_parent_snapshot(
+                manifest=manifest,
+                parent_snapshot=parent_snapshot,
+                parent_snapshot_id=parent_snapshot_id,
+                context=context,
+            )
 
         # Update history
         new_classes = [v.type for v in snapshot.truth_graph.vulns]
@@ -90,3 +136,529 @@ class Mutator:
     def history(self) -> list[str]:
         """All vuln classes used so far, in order."""
         return list(self._history)
+
+    def _hydrate_root_snapshot(
+        self,
+        snapshot: SnapshotSpec,
+        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.lineage = LineageMetadata(
+            manifest_id=str(manifest.get("name", "")),
+            generation_depth=0,
+            mutation_summary=["compile_base_snapshot"],
+        )
+        root.mutation_plan = None
+        return root
+
+    def _mutate_parent_snapshot(
+        self,
+        *,
+        manifest: dict[str, Any],
+        parent_snapshot: SnapshotSpec,
+        parent_snapshot_id: str | None,
+        context: BuildContext,
+    ) -> SnapshotSpec:
+        rng = random.Random(context.seed if context.seed is not None else self._episode_count + 1)
+        child = parent_snapshot.model_copy(deep=True)
+        child.topology = _ensure_mutable_topology(child.topology, manifest)
+
+        plan = self._plan_mutations(
+            manifest=manifest,
+            snapshot=child,
+            parent_snapshot_id=parent_snapshot_id,
+            context=context,
+            rng=rng,
+        )
+        self._apply_plan(child, plan, manifest, context)
+
+        lineage = parent_snapshot.lineage.model_copy(deep=True)
+        child.lineage = LineageMetadata(
+            parent_snapshot_id=parent_snapshot_id or parent_snapshot.lineage.snapshot_id or None,
+            root_snapshot_id=lineage.root_snapshot_id or parent_snapshot_id or "",
+            manifest_id=lineage.manifest_id or str(manifest.get("name", "")),
+            generation_depth=lineage.generation_depth + 1,
+            mutation_ids=[op.mutation_id for op in plan.ops],
+            mutation_summary=[_mutation_summary(op) for op in plan.ops],
+        )
+        child.mutation_plan = plan
+        return child
+
+    def _plan_mutations(
+        self,
+        *,
+        manifest: dict[str, Any],
+        snapshot: SnapshotSpec,
+        parent_snapshot_id: str | None,
+        context: BuildContext,
+        rng: random.Random,
+    ) -> MutationPlan:
+        ops: list[MutationOp] = []
+        used_ids: set[str] = set()
+
+        structural_candidates = []
+        op = self._candidate_add_service(manifest, snapshot, rng)
+        if op is not None:
+            structural_candidates.append(op)
+        op = self._candidate_add_user(manifest, snapshot, context, rng)
+        if op is not None:
+            structural_candidates.append(op)
+        op = self._candidate_add_dependency_edge(manifest, snapshot, rng)
+        if op is not None:
+            structural_candidates.append(op)
+        op = self._candidate_add_trust_edge(manifest, snapshot, rng)
+        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:
+            security_candidates.append(op)
+        op = self._candidate_add_benign_noise(snapshot, rng)
+        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)
+
+        if not ops:
+            fallback = self._candidate_add_benign_noise(snapshot, rng)
+            if fallback is not None:
+                ops.append(fallback)
+
+        return MutationPlan(
+            parent_snapshot_id=parent_snapshot_id,
+            ops=ops,
+            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),
+        )
+
+    def _candidate_add_service(
+        self,
+        manifest: dict[str, Any],
+        snapshot: SnapshotSpec,
+        rng: random.Random,
+    ) -> MutationOp | None:
+        topology = snapshot.topology
+        host_catalog = topology.get("host_catalog", {})
+        host_details = topology.get("host_details", {})
+        candidates: list[tuple[str, str]] = []
+        if not isinstance(host_catalog, dict) or not isinstance(host_details, dict):
+            return None
+        for host, raw_catalog in host_catalog.items():
+            if not isinstance(raw_catalog, dict):
+                continue
+            allowed = raw_catalog.get("services", [])
+            detail = host_details.get(host, {})
+            current = detail.get("services", []) if isinstance(detail, dict) else []
+            if not isinstance(allowed, list) or not isinstance(current, list):
+                continue
+            for service in allowed:
+                if service and service not in current:
+                    candidates.append((str(host), str(service)))
+        if not candidates:
+            return None
+        host, service = rng.choice(candidates)
+        return MutationOp(
+            mutation_id=f"mut_add_service_{host}_{service}",
+            op_type="add_service",
+            target_selector={"host": host},
+            params={"service": service},
+            expected_effects=[f"service {service} added to {host}"],
+            risk_tags=["surface_expansion"],
+        )
+
+    def _candidate_add_user(
+        self,
+        manifest: dict[str, Any],
+        snapshot: SnapshotSpec,
+        context: BuildContext,
+        rng: random.Random,
+    ) -> MutationOp | None:
+        existing = _existing_usernames(snapshot)
+        candidates = [
+            raw for raw in manifest.get("users", [])
+            if isinstance(raw, dict) and raw.get("username") not in existing
+        ]
+        if not candidates:
+            return None
+        user = deepcopy(rng.choice(candidates))
+        username = str(user.get("username", "")).strip()
+        if not username:
+            return None
+        password = _predictable_password(username, context.seed)
+        return MutationOp(
+            mutation_id=f"mut_add_user_{username}",
+            op_type="add_user",
+            target_selector={"user": username},
+            params={
+                "username": username,
+                "password": password,
+                "hosts": deepcopy(user.get("hosts", [])),
+                "groups": [str(user.get("department", "") or "users").lower().replace(" ", "_")],
+                "email": str(user.get("email", "")),
+                "full_name": str(user.get("full_name", "")),
+                "department": str(user.get("department", "")),
+                "role": str(user.get("role", "")),
+            },
+            expected_effects=[f"user {username} added to snapshot accounts"],
+            risk_tags=["identity_expansion"],
+        )
+
+    def _candidate_add_dependency_edge(
+        self,
+        manifest: dict[str, Any],
+        snapshot: SnapshotSpec,
+        rng: random.Random,
+    ) -> MutationOp | None:
+        topology = snapshot.topology
+        current = {
+            (str(edge.get("source", "")), str(edge.get("target", "")))
+            for edge in topology.get("dependency_edges", [])
+            if isinstance(edge, dict)
+        }
+        candidates: list[tuple[str, str]] = []
+        for raw in manifest.get("topology", {}).get("hosts", []):
+            if not isinstance(raw, dict):
+                continue
+            source = str(raw.get("name", "")).strip()
+            raw_targets = raw.get("connects_to", [])
+            if not source or not isinstance(raw_targets, list):
+                continue
+            for target_raw in raw_targets:
+                target = str(target_raw).strip()
+                if target and (source, target) not in current:
+                    candidates.append((source, target))
+        if not candidates:
+            return None
+        source, target = rng.choice(candidates)
+        return MutationOp(
+            mutation_id=f"mut_add_dep_{source}_{target}",
+            op_type="add_dependency_edge",
+            target_selector={"source": source, "target": target},
+            params={},
+            expected_effects=[f"dependency edge {source}->{target} added"],
+            risk_tags=["topology_expansion"],
+        )
+
+    def _candidate_add_trust_edge(
+        self,
+        manifest: dict[str, Any],
+        snapshot: SnapshotSpec,
+        rng: random.Random,
+    ) -> MutationOp | None:
+        topology = snapshot.topology
+        current = {
+            (
+                str(edge.get("source", "")),
+                str(edge.get("target", "")),
+                str(edge.get("type", "")),
+            )
+            for edge in topology.get("trust_edges", [])
+            if isinstance(edge, dict)
+        }
+        candidates: list[dict[str, str]] = []
+        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 source and target and (source, target, edge_type) not in current:
+                candidates.append(
+                    {
+                        "source": source,
+                        "target": target,
+                        "type": edge_type,
+                        "context": str(raw.get("context") or raw.get("description") or ""),
+                    }
+                )
+        if not candidates:
+            return None
+        choice = rng.choice(candidates)
+        return MutationOp(
+            mutation_id=f"mut_add_trust_{choice['source']}_{choice['target']}_{choice['type']}",
+            op_type="add_trust_edge",
+            target_selector={"source": choice["source"], "target": choice["target"]},
+            params={"type": choice["type"], "context": choice["context"]},
+            expected_effects=[f"trust edge {choice['source']}->{choice['target']} added"],
+            risk_tags=["trust_expansion"],
+        )
+
+    def _candidate_seed_vuln(
+        self,
+        manifest: dict[str, Any],
+        snapshot: SnapshotSpec,
+        context: BuildContext,
+        rng: random.Random,
+    ) -> MutationOp | None:
+        allowed = [str(v) for v in manifest.get("bug_families", []) if v]
+        if not allowed:
+            return None
+        existing = {v.type for v in snapshot.truth_graph.vulns}
+        preferred = [v for v in context.weak_areas if v in allowed and v not in existing]
+        remaining = [v for v in allowed if v not in existing]
+        choices = preferred or remaining or allowed
+        vuln_type = rng.choice(choices)
+
+        host_catalog = snapshot.topology.get("host_catalog", {})
+        host_candidates = list(host_catalog.keys()) if isinstance(host_catalog, dict) else []
+        if not host_candidates:
+            host_candidates = list(_existing_hosts(snapshot))
+        if not host_candidates:
+            return None
+        host = str(rng.choice(host_candidates))
+        service = ""
+        if isinstance(host_catalog, dict):
+            raw_catalog = host_catalog.get(host, {})
+            if isinstance(raw_catalog, dict):
+                raw_services = raw_catalog.get("services", [])
+                if isinstance(raw_services, list) and raw_services:
+                    service = str(raw_services[0])
+
+        return MutationOp(
+            mutation_id=f"mut_seed_vuln_{vuln_type}_{host}_{len(snapshot.truth_graph.vulns)+1}",
+            op_type="seed_vuln",
+            target_selector={"host": host},
+            params={"vuln_type": vuln_type, "service": service},
+            expected_effects=[f"new {vuln_type} foothold on {host}"],
+            risk_tags=["security_condition"],
+        )
+
+    def _candidate_add_benign_noise(
+        self,
+        snapshot: SnapshotSpec,
+        rng: random.Random,
+    ) -> MutationOp | None:
+        locations = [item.location for item in snapshot.evidence_spec if item.location]
+        location = rng.choice(locations) if locations else "siem:background.log"
+        return MutationOp(
+            mutation_id=f"mut_add_noise_{len(snapshot.evidence_spec)+1}",
+            op_type="add_benign_noise",
+            target_selector={"location": location},
+            params={"location": location},
+            expected_effects=[f"benign evidence noise added at {location}"],
+            risk_tags=["observability_noise"],
+        )
+
+    def _apply_plan(
+        self,
+        snapshot: SnapshotSpec,
+        plan: MutationPlan,
+        manifest: dict[str, Any],
+        context: BuildContext,
+    ) -> None:
+        topology = snapshot.topology
+        host_details = topology.setdefault("host_details", {})
+        dependency_edges = topology.setdefault("dependency_edges", [])
+        trust_edges = topology.setdefault("trust_edges", [])
+        users = topology.setdefault("users", [])
+
+        if not isinstance(host_details, dict):
+            host_details = {}
+            topology["host_details"] = host_details
+        if not isinstance(dependency_edges, list):
+            dependency_edges = []
+            topology["dependency_edges"] = dependency_edges
+        if not isinstance(trust_edges, list):
+            trust_edges = []
+            topology["trust_edges"] = trust_edges
+        if not isinstance(users, list):
+            users = []
+            topology["users"] = users
+
+        for op in plan.ops:
+            if op.op_type not in _SUPPORTED_MUTATION_OPS:
+                raise ValueError(f"Unsupported mutation op {op.op_type!r}")
+
+            if op.op_type == "add_service":
+                host = op.target_selector["host"]
+                detail = host_details.setdefault(host, {"services": [], "connects_to": []})
+                services = detail.setdefault("services", [])
+                service = str(op.params.get("service", "")).strip()
+                if service and service not in services:
+                    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", "")),
+                    }
+                )
+
+            elif op.op_type == "add_dependency_edge":
+                dependency_edges.append(
+                    {
+                        "source": op.target_selector["source"],
+                        "target": op.target_selector["target"],
+                    }
+                )
+
+            elif op.op_type == "add_trust_edge":
+                trust_edges.append(
+                    {
+                        "source": op.target_selector["source"],
+                        "target": op.target_selector["target"],
+                        "type": str(op.params.get("type", "")),
+                        "context": str(op.params.get("context", "")),
+                    }
+                )
+
+            elif op.op_type == "seed_vuln":
+                vuln_type = str(op.params.get("vuln_type", "")).strip()
+                host = op.target_selector["host"]
+                service = str(op.params.get("service", "")).strip()
+                vuln_id = f"{vuln_type}_{len(snapshot.truth_graph.vulns) + 1}"
+                snapshot.truth_graph.vulns.append(
+                    Vulnerability(
+                        id=vuln_id,
+                        type=vuln_type,
+                        host=host,
+                        service=service,
+                        injection_point=_INJECTION_POINTS.get(vuln_type, f"/debug/{vuln_type}"),
+                        vulnerable_code=f"// mutation-added {vuln_type} surface on {host}",
+                        root_cause=f"Mutation introduced {vuln_type} on {host}",
+                        blast_radius=f"Additional foothold on {host}",
+                        remediation=f"Remove the {vuln_type} issue and review dependent trust paths",
+                    )
+                )
+                snapshot.truth_graph.exploit_chain.append(
+                    ExploitStep(
+                        vuln_id=vuln_id,
+                        command=f"probe {host} for {vuln_type}",
+                        description=f"Use the new {vuln_type} foothold on {host}",
+                    )
+                )
+                snapshot.evidence_spec.append(
+                    EvidenceItem(
+                        type="log_entry",
+                        location=f"{host}:app.log",
+                        pattern=f"Mutation-added {vuln_type} activity on {host}",
+                    )
+                )
+
+            elif op.op_type == "add_benign_noise":
+                location = str(op.params.get("location", "siem:background.log"))
+                snapshot.evidence_spec.append(
+                    EvidenceItem(
+                        type="log_entry",
+                        location=location,
+                        pattern=(
+                            f"Benign background activity {context.episode_count + len(snapshot.evidence_spec)}"
+                        ),
+                    )
+                )
+
+        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
+
+
+def _existing_hosts(snapshot: SnapshotSpec) -> set[str]:
+    hosts: set[str] = set()
+    for raw in snapshot.topology.get("hosts", []):
+        if isinstance(raw, dict):
+            name = str(raw.get("name", "")).strip()
+            if name:
+                hosts.add(name)
+        else:
+            name = str(raw).strip()
+            if name:
+                hosts.add(name)
+    return hosts
+
+
+def _existing_usernames(snapshot: SnapshotSpec) -> set[str]:
+    usernames: set[str] = set()
+    for raw in snapshot.topology.get("users", []):
+        if not isinstance(raw, dict):
+            continue
+        username = str(raw.get("username", "")).strip()
+        if username:
+            usernames.add(username)
+    return usernames
+
+
+def _predictable_password(username: str, seed: int | None) -> str:
+    suffix = 2025 if seed is None else 2025 + (seed % 3)
+    base = username.split("@", 1)[0] or "Welcome"
+    return f"{base.capitalize()}!{suffix}"
+
+
+def _mutation_summary(op: MutationOp) -> str:
+    if op.op_type == "add_service":
+        return f"add service {op.params.get('service', '')} to {op.target_selector.get('host', '')}"
+    if op.op_type == "add_user":
+        return f"add user {op.params.get('username', '')}"
+    if op.op_type == "add_dependency_edge":
+        return (
+            f"add dependency {op.target_selector.get('source', '')}->"
+            f"{op.target_selector.get('target', '')}"
+        )
+    if op.op_type == "add_trust_edge":
+        return (
+            f"add trust {op.target_selector.get('source', '')}->"
+            f"{op.target_selector.get('target', '')}"
+        )
+    if op.op_type == "seed_vuln":
+        return (
+            f"seed {op.params.get('vuln_type', '')} on "
+            f"{op.target_selector.get('host', '')}"
+        )
+    if op.op_type == "add_benign_noise":
+        return f"add benign noise at {op.params.get('location', '')}"
+    return op.op_type

src/open_range/builder/snapshot_store.py

@@ -70,6 +70,10 @@ class SnapshotStore:
             "flag_count": len(snapshot.flags),
             "npc_count": len(snapshot.npc_personas),
             "has_compose": bool(snapshot.compose),
+            "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_path = snap_dir / "metadata.json"

src/open_range/protocols.py

@@ -46,6 +46,40 @@ class BuildContext(BaseModel):
     )
 
 
+class MutationOp(BaseModel):
+    """Single typed edit applied to derive a child snapshot from a parent."""
+
+    mutation_id: str
+    op_type: str
+    target_selector: dict[str, str] = Field(default_factory=dict)
+    magnitude: int = 1
+    params: dict[str, Any] = Field(default_factory=dict)
+    expected_effects: list[str] = Field(default_factory=list)
+    risk_tags: list[str] = Field(default_factory=list)
+
+
+class MutationPlan(BaseModel):
+    """Ordered list of mutations used to produce a child snapshot."""
+
+    parent_snapshot_id: str | None = None
+    ops: list[MutationOp] = Field(default_factory=list)
+    predicted_complexity_delta: int = 0
+    predicted_chain_delta: int = 0
+    predicted_novelty: float = 0.0
+
+
+class LineageMetadata(BaseModel):
+    """Lineage and mutation provenance for a stored snapshot."""
+
+    snapshot_id: str = ""
+    parent_snapshot_id: str | None = None
+    root_snapshot_id: str = ""
+    manifest_id: str = ""
+    generation_depth: int = 0
+    mutation_ids: list[str] = Field(default_factory=list)
+    mutation_summary: list[str] = Field(default_factory=list)
+
+
 class Vulnerability(BaseModel):
     """Single planted vulnerability in the truth graph."""
 
@@ -160,6 +194,8 @@ class SnapshotSpec(BaseModel):
     task: TaskSpec = Field(default_factory=TaskSpec)
     compose: dict[str, Any] = Field(default_factory=dict)  # rendered docker-compose
     files: dict[str, str] = Field(default_factory=dict)  # path -> content
+    lineage: LineageMetadata = Field(default_factory=LineageMetadata)
+    mutation_plan: MutationPlan | None = None
 
 
 class Stimulus(BaseModel):

src/open_range/server/runtime.py

@@ -32,6 +32,8 @@ from open_range.protocols import (
     SnapshotSpec,
 )
 from open_range.server.models import RangeState
+from open_range.validator.graph_consistency import GraphConsistencyCheck
+from open_range.validator.manifest_compliance import ManifestComplianceCheck
 from open_range.validator.task_feasibility import TaskFeasibilityCheck
 from open_range.validator.validator import ValidationResult, ValidatorGate
 
@@ -242,11 +244,13 @@ def _default_builder() -> SnapshotBuilder:
     )
 
 
-def _default_validator() -> ValidatorGate:
+def _default_validator(manifest: dict[str, Any]) -> ValidatorGate:
     # These checks work directly against the compiled snapshot spec and do not
     # require booted containers. They are the safe default for shipped mode.
     return ValidatorGate(
         [
+            ManifestComplianceCheck(manifest),
+            GraphConsistencyCheck(),
             StructuralSnapshotCheck(),
             TaskFeasibilityCheck(),
         ]
@@ -280,7 +284,7 @@ class ManagedSnapshotRuntime:
         self.store = SnapshotStore(str(self.store_dir))
         self.builder = builder or _default_builder()
         self.mutator = Mutator(self.builder)
-        self.validator = validator or _default_validator()
+        self.validator = validator or _default_validator(self.manifest)
         self.renderer = SnapshotRenderer()
         self.curriculum = CurriculumTracker()
         self.pool_size = max(1, pool_size)
@@ -452,11 +456,14 @@ class ManagedSnapshotRuntime:
         last_error: str | None = None
         for attempt in range(1, self.generation_retries + 1):
             context = self._build_context()
+            parent_entry = self._select_parent_entry()
             snapshot = _run_coro_sync(
                 self.mutator.mutate(
                     self.manifest,
                     context=context,
                     error={"message": last_error} if last_error else None,
+                    parent_snapshot=parent_entry.snapshot if parent_entry else None,
+                    parent_snapshot_id=parent_entry.snapshot_id if parent_entry else None,
                 )
             )
             validation = self._validate_snapshot(snapshot)
@@ -516,6 +523,11 @@ 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):
+        if self.snapshot_count() == 0:
+            return None
+        return _run_coro_sync(self.store.select_entry(strategy=self.selection_strategy))
+
     def _snapshot_dir(self, snapshot_id: str) -> Path:
         return self.store_dir / snapshot_id
 
@@ -532,6 +544,9 @@ class ManagedSnapshotRuntime:
         topology = dict(rendered.topology)
         topology["snapshot_id"] = snapshot_id
         rendered.topology = topology
+        rendered.lineage.snapshot_id = snapshot_id
+        if not rendered.lineage.root_snapshot_id:
+            rendered.lineage.root_snapshot_id = snapshot_id
 
         snapshot_dir = self._snapshot_dir(snapshot_id)
         artifacts_dir = self._artifacts_dir(snapshot_id)

src/open_range/validator/graph_consistency.py

@@ -0,0 +1,73 @@
+"""Graph-level consistency checks for compiled snapshot state."""
+
+from __future__ import annotations
+
+from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
+from open_range.validator.graphs import compile_snapshot_graphs
+
+
+class GraphConsistencyCheck:
+    """Verify internal consistency of the canonical graph views."""
+
+    async def check(self, snapshot: SnapshotSpec, containers: ContainerSet) -> CheckResult:
+        compiled = compile_snapshot_graphs(snapshot)
+        issues: list[str] = []
+
+        for source, target in compiled.dependency_edges:
+            if source not in compiled.hosts or target not in compiled.hosts:
+                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")
+
+        lineage = snapshot.lineage
+        if lineage.generation_depth == 0 and lineage.parent_snapshot_id:
+            issues.append("root snapshot must not have parent_snapshot_id")
+        if lineage.generation_depth > 0 and not lineage.parent_snapshot_id:
+            issues.append("child snapshot missing parent_snapshot_id")
+        if snapshot.mutation_plan is not None:
+            if snapshot.mutation_plan.parent_snapshot_id != lineage.parent_snapshot_id:
+                issues.append("mutation plan parent does not match lineage parent")
+            for op in snapshot.mutation_plan.ops:
+                if op.op_type in {"add_service", "seed_vuln"}:
+                    host = op.target_selector.get("host", "")
+                    if host and host not in compiled.hosts:
+                        issues.append(
+                            f"mutation '{op.mutation_id}' targets unknown host '{host}'"
+                        )
+                if op.op_type == "add_dependency_edge":
+                    source = op.target_selector.get("source", "")
+                    target = op.target_selector.get("target", "")
+                    if source and source not in compiled.hosts:
+                        issues.append(
+                            f"mutation '{op.mutation_id}' source host '{source}' missing"
+                        )
+                    if target and target not in compiled.hosts:
+                        issues.append(
+                            f"mutation '{op.mutation_id}' target host '{target}' missing"
+                        )
+                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:
+                        issues.append(
+                            f"mutation '{op.mutation_id}' source user '{source}' missing"
+                        )
+                    if target and target not in compiled.users:
+                        issues.append(
+                            f"mutation '{op.mutation_id}' target user '{target}' missing"
+                        )
+
+        passed = len(issues) == 0
+        return CheckResult(
+            name="graph_consistency",
+            passed=passed,
+            details={
+                "hosts": len(compiled.hosts),
+                "users": len(compiled.users),
+                "dependency_edges": len(compiled.dependency_edges),
+                "trust_edges": len(compiled.trust_edges),
+            },
+            error="" if passed else "; ".join(issues),
+        )

src/open_range/validator/graphs.py

@@ -0,0 +1,121 @@
+"""Compile SnapshotSpec into lightweight canonical graph views.
+
+These helpers intentionally stay small and dependency-free. The validator uses
+them to reason about host membership, dependency edges, trust edges, evidence
+locations, and mutation targets before any live container checks run.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from open_range.protocols import SnapshotSpec
+
+
+@dataclass(frozen=True, slots=True)
+class CompiledGraphs:
+    """Canonical graph-like views derived from a snapshot."""
+
+    hosts: frozenset[str]
+    users: frozenset[str]
+    services_by_host: dict[str, frozenset[str]]
+    dependency_edges: frozenset[tuple[str, str]]
+    trust_edges: frozenset[tuple[str, str, str]]
+    vuln_ids: frozenset[str]
+    evidence_locations: frozenset[str]
+
+
+def compile_snapshot_graphs(snapshot: SnapshotSpec) -> CompiledGraphs:
+    """Compile a snapshot into canonical graph views."""
+
+    topology = snapshot.topology or {}
+    hosts = _compile_hosts(topology)
+    users = _compile_users(topology)
+    services_by_host = _compile_services(topology, hosts)
+    dependency_edges = _compile_dependency_edges(topology)
+    trust_edges = _compile_trust_edges(topology)
+    vuln_ids = frozenset(v.id for v in snapshot.truth_graph.vulns if v.id)
+    evidence_locations = frozenset(item.location for item in snapshot.evidence_spec if item.location)
+
+    return CompiledGraphs(
+        hosts=hosts,
+        users=users,
+        services_by_host=services_by_host,
+        dependency_edges=dependency_edges,
+        trust_edges=trust_edges,
+        vuln_ids=vuln_ids,
+        evidence_locations=evidence_locations,
+    )
+
+
+def _compile_hosts(topology: dict[str, object]) -> frozenset[str]:
+    raw_hosts = topology.get("hosts", [])
+    hosts: set[str] = set()
+    for raw in raw_hosts if isinstance(raw_hosts, list) else []:
+        if isinstance(raw, dict):
+            name = str(raw.get("name", "")).strip()
+            if name:
+                hosts.add(name)
+        else:
+            name = str(raw).strip()
+            if name:
+                hosts.add(name)
+    return frozenset(hosts)
+
+
+def _compile_users(topology: dict[str, object]) -> frozenset[str]:
+    raw_users = topology.get("users", [])
+    users: set[str] = set()
+    for raw in raw_users if isinstance(raw_users, list) else []:
+        if not isinstance(raw, dict):
+            continue
+        username = str(raw.get("username", "")).strip()
+        if username:
+            users.add(username)
+    return frozenset(users)
+
+
+def _compile_services(
+    topology: dict[str, object],
+    hosts: frozenset[str],
+) -> dict[str, frozenset[str]]:
+    host_details = topology.get("host_details", {})
+    compiled: dict[str, frozenset[str]] = {}
+    for host in hosts:
+        detail = {}
+        if isinstance(host_details, dict):
+            raw_detail = host_details.get(host, {})
+            if isinstance(raw_detail, dict):
+                detail = raw_detail
+        services = detail.get("services", [])
+        if not isinstance(services, list):
+            services = []
+        compiled[host] = frozenset(str(service) for service in services if service)
+    return compiled
+
+
+def _compile_dependency_edges(topology: dict[str, object]) -> frozenset[tuple[str, str]]:
+    raw_edges = topology.get("dependency_edges", [])
+    edges: set[tuple[str, str]] = set()
+    for raw in raw_edges if isinstance(raw_edges, list) else []:
+        if not isinstance(raw, dict):
+            continue
+        source = str(raw.get("source", "")).strip()
+        target = str(raw.get("target", "")).strip()
+        if source and target:
+            edges.add((source, target))
+    return frozenset(edges)
+
+
+def _compile_trust_edges(topology: dict[str, object]) -> frozenset[tuple[str, str, str]]:
+    raw_edges = topology.get("trust_edges", [])
+    edges: set[tuple[str, str, str]] = set()
+    for raw in raw_edges if isinstance(raw_edges, list) else []:
+        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 source and target:
+            edges.add((source, target, edge_type))
+    return frozenset(edges)

src/open_range/validator/manifest_compliance.py

@@ -0,0 +1,183 @@
+"""Manifest-bounded legality checks for candidate snapshots."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
+from open_range.validator.graphs import compile_snapshot_graphs
+
+_SUPPORTED_MUTATION_OPS = {
+    "add_service",
+    "add_user",
+    "add_dependency_edge",
+    "add_trust_edge",
+    "seed_vuln",
+    "add_benign_noise",
+}
+
+_SYSTEM_USERS = {"admin", "testuser"}
+
+
+class ManifestComplianceCheck:
+    """Ensure a candidate child stays inside the manifest-defined family."""
+
+    def __init__(self, manifest: dict[str, Any]) -> None:
+        self.manifest = manifest
+
+    async def check(self, snapshot: SnapshotSpec, containers: ContainerSet) -> CheckResult:
+        compiled = compile_snapshot_graphs(snapshot)
+        issues: list[str] = []
+
+        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_services = _manifest_services(self.manifest)
+        allowed_dependency_edges = _manifest_dependency_edges(self.manifest)
+        allowed_trust_edges = _manifest_trust_edges(self.manifest)
+
+        unknown_hosts = compiled.hosts - manifest_hosts
+        if unknown_hosts:
+            issues.append(f"hosts outside manifest family: {sorted(unknown_hosts)}")
+
+        illegal_users = {
+            user
+            for user in compiled.users
+            if user not in allowed_users and user not in _SYSTEM_USERS and not user.startswith("svc_")
+        }
+        if illegal_users:
+            issues.append(f"users outside manifest family: {sorted(illegal_users)}")
+
+        for host, services in compiled.services_by_host.items():
+            illegal = services - allowed_services.get(host, frozenset())
+            if illegal:
+                issues.append(
+                    f"host '{host}' has services outside manifest family: {sorted(illegal)}"
+                )
+
+        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}'")
+            if vuln.host and vuln.host not in manifest_hosts:
+                issues.append(f"vuln '{vuln.id}' references host outside manifest '{vuln.host}'")
+
+        plan = snapshot.mutation_plan
+        if plan is not None:
+            for op in plan.ops:
+                if op.op_type not in _SUPPORTED_MUTATION_OPS:
+                    issues.append(f"unsupported mutation op '{op.op_type}'")
+                    continue
+
+                if op.op_type == "add_service":
+                    host = op.target_selector.get("host", "")
+                    service = str(op.params.get("service", "")).strip()
+                    if host not in manifest_hosts:
+                        issues.append(f"add_service targets unknown host '{host}'")
+                    elif service and service not in allowed_services.get(host, frozenset()):
+                        issues.append(f"add_service introduces illegal service '{service}' on '{host}'")
+
+                if op.op_type == "add_user":
+                    username = str(op.params.get("username", "")).strip()
+                    if username and username not in allowed_users:
+                        issues.append(f"add_user introduces unknown manifest user '{username}'")
+
+                if op.op_type == "add_dependency_edge":
+                    source = op.target_selector.get("source", "")
+                    target = op.target_selector.get("target", "")
+                    if (source, target) not in allowed_dependency_edges:
+                        issues.append(
+                            f"add_dependency_edge introduces illegal edge '{source}->{target}'"
+                        )
+
+                if op.op_type == "add_trust_edge":
+                    source = op.target_selector.get("source", "")
+                    target = op.target_selector.get("target", "")
+                    edge_type = str(op.params.get("type", "")).strip()
+                    if (source, target, edge_type) not in allowed_trust_edges:
+                        issues.append(
+                            f"add_trust_edge introduces illegal trust edge "
+                            f"'{source}->{target}' ({edge_type})"
+                        )
+
+                if op.op_type == "seed_vuln":
+                    host = op.target_selector.get("host", "")
+                    vuln_type = str(op.params.get("vuln_type", "")).strip()
+                    if host not in manifest_hosts:
+                        issues.append(f"seed_vuln targets unknown host '{host}'")
+                    if vuln_type and vuln_type not in allowed_bug_families:
+                        issues.append(f"seed_vuln uses illegal family '{vuln_type}'")
+
+        passed = len(issues) == 0
+        return CheckResult(
+            name="manifest_compliance",
+            passed=passed,
+            details={
+                "issue_count": len(issues),
+                "manifest": self.manifest.get("name", ""),
+            },
+            error="" if passed else "; ".join(issues),
+        )
+
+
+def _manifest_hosts(manifest: dict[str, Any]) -> set[str]:
+    hosts: set[str] = set()
+    for raw in manifest.get("topology", {}).get("hosts", []):
+        if isinstance(raw, dict):
+            name = str(raw.get("name", "")).strip()
+            if name:
+                hosts.add(name)
+    return hosts
+
+
+def _manifest_users(manifest: dict[str, Any]) -> set[str]:
+    users: set[str] = set()
+    for raw in manifest.get("users", []):
+        if isinstance(raw, dict):
+            username = str(raw.get("username", "")).strip()
+            if username:
+                users.add(username)
+    return users
+
+
+def _manifest_services(manifest: dict[str, Any]) -> dict[str, frozenset[str]]:
+    services: dict[str, frozenset[str]] = {}
+    for raw in manifest.get("topology", {}).get("hosts", []):
+        if not isinstance(raw, dict):
+            continue
+        name = str(raw.get("name", "")).strip()
+        if not name:
+            continue
+        raw_services = raw.get("services", [])
+        if not isinstance(raw_services, list):
+            raw_services = []
+        services[name] = frozenset(str(service) for service in raw_services if service)
+    return services
+
+
+def _manifest_dependency_edges(manifest: dict[str, Any]) -> set[tuple[str, str]]:
+    edges: set[tuple[str, str]] = set()
+    for raw in manifest.get("topology", {}).get("hosts", []):
+        if not isinstance(raw, dict):
+            continue
+        source = str(raw.get("name", "")).strip()
+        raw_targets = raw.get("connects_to", [])
+        if not source or not isinstance(raw_targets, list):
+            continue
+        for raw_target in raw_targets:
+            target = str(raw_target).strip()
+            if target:
+                edges.add((source, target))
+    return edges
+
+
+def _manifest_trust_edges(manifest: dict[str, Any]) -> set[tuple[str, str, str]]:
+    edges: set[tuple[str, str, str]] = set()
+    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 source and target:
+            edges.add((source, target, edge_type))
+    return edges

tests/test_builder.py

@@ -103,6 +103,28 @@ async def test_template_builder_has_task_briefings(tier1_manifest):
     assert spec.task.blue_briefing != ""
 
 
+@pytest.mark.asyncio
+async def test_mutator_builds_child_snapshot_with_lineage(tier1_manifest):
+    from open_range.builder.builder import TemplateOnlyBuilder
+    from open_range.builder.mutator import Mutator
+
+    mutator = Mutator(TemplateOnlyBuilder())
+    root = await mutator.mutate(tier1_manifest, context=BuildContext(seed=1, tier=1))
+    child = await mutator.mutate(
+        tier1_manifest,
+        context=BuildContext(seed=2, tier=1),
+        parent_snapshot=root,
+        parent_snapshot_id="root_snap",
+    )
+
+    assert child.lineage.parent_snapshot_id == "root_snap"
+    assert child.lineage.generation_depth == 1
+    assert child.mutation_plan is not None
+    assert child.mutation_plan.parent_snapshot_id == "root_snap"
+    assert child.mutation_plan.ops
+    assert child.lineage.mutation_summary
+
+
 # ---------------------------------------------------------------------------
 # FileBuilder
 # ---------------------------------------------------------------------------

tests/test_runtime.py

@@ -80,6 +80,42 @@ class TestManagedSnapshotRuntime:
         finally:
             runtime.stop()
 
+    def test_start_records_root_and_child_lineage(self, tier1_manifest, tmp_path):
+        runtime = ManagedSnapshotRuntime(
+            manifest=tier1_manifest,
+            store_dir=tmp_path / "snapshots",
+            pool_size=2,
+            selection_strategy="latest",
+            refill_enabled=False,
+        )
+
+        runtime.start()
+        try:
+            listing = runtime.list_snapshots()
+            assert len(listing) == 2
+            depths = {item["generation_depth"] for item in listing}
+            assert 0 in depths
+            assert 1 in depths
+            assert any(item["parent_snapshot_id"] for item in listing)
+        finally:
+            runtime.stop()
+
+    def test_acquire_snapshot_exposes_lineage_metadata(self, tier1_manifest, tmp_path):
+        runtime = ManagedSnapshotRuntime(
+            manifest=tier1_manifest,
+            store_dir=tmp_path / "snapshots",
+            pool_size=2,
+            refill_enabled=False,
+        )
+
+        runtime.start()
+        try:
+            admitted = runtime.acquire_snapshot()
+            assert admitted.snapshot.lineage.snapshot_id == admitted.snapshot_id
+            assert admitted.snapshot.lineage.root_snapshot_id
+        finally:
+            runtime.stop()
+
 
 class TestEnvironmentRuntimeIntegration:
     def test_reset_uses_managed_runtime_snapshot(self, tier1_manifest, tmp_path):

tests/test_validator.py

@@ -10,6 +10,8 @@ from open_range.protocols import (
     EvidenceItem,
     FlagSpec,
     GoldenPathStep,
+    MutationOp,
+    MutationPlan,
     NPCPersona,
     SnapshotSpec,
     TaskSpec,
@@ -19,6 +21,58 @@ from open_range.protocols import (
 from open_range.validator.validator import ValidatorGate, ValidationResult
 
 
+@pytest.mark.asyncio
+async def test_manifest_compliance_rejects_illegal_mutation_plan(
+    tier1_manifest,
+    sample_snapshot_spec,
+    mock_containers,
+):
+    from open_range.validator.manifest_compliance import ManifestComplianceCheck
+
+    spec = sample_snapshot_spec.model_copy(deep=True)
+    spec.mutation_plan = MutationPlan(
+        parent_snapshot_id="root_snap",
+        ops=[
+            MutationOp(
+                mutation_id="illegal1",
+                op_type="seed_vuln",
+                target_selector={"host": "web"},
+                params={"vuln_type": "totally_fake_bug"},
+            )
+        ],
+    )
+    spec.lineage.parent_snapshot_id = "root_snap"
+    spec.lineage.generation_depth = 1
+
+    result = await ManifestComplianceCheck(tier1_manifest).check(spec, mock_containers)
+    assert result.passed is False
+    assert "illegal family" in result.error
+
+
+@pytest.mark.asyncio
+async def test_graph_consistency_rejects_missing_parent_lineage(sample_snapshot_spec, mock_containers):
+    from open_range.validator.graph_consistency import GraphConsistencyCheck
+
+    spec = sample_snapshot_spec.model_copy(deep=True)
+    spec.mutation_plan = MutationPlan(
+        parent_snapshot_id="root_snap",
+        ops=[
+            MutationOp(
+                mutation_id="mut1",
+                op_type="add_benign_noise",
+                target_selector={"location": "siem:noise.log"},
+                params={"location": "siem:noise.log"},
+            )
+        ],
+    )
+    spec.lineage.generation_depth = 1
+    spec.lineage.parent_snapshot_id = None
+
+    result = await GraphConsistencyCheck().check(spec, mock_containers)
+    assert result.passed is False
+    assert "missing parent_snapshot_id" in result.error
+
+
 # ---------------------------------------------------------------------------
 # Check 1: BuildBoot
 # ---------------------------------------------------------------------------