Isolate runtime-backed snapshots across resets

src/open_range/server/compose_runner.py

@@ -47,9 +47,10 @@ class ComposeProjectRunner:
         snapshot_id: str,
         artifacts_dir: Path,
         compose: dict[str, Any],
+        project_name: str | None = None,
     ) -> BootedSnapshotProject:
         compose_file = artifacts_dir / "docker-compose.yml"
-        project_name = self.project_name_for(snapshot_id)
+        project_name = project_name or self.project_name_for(snapshot_id)
 
         self._run(
             [

src/open_range/server/environment.py

@@ -28,6 +28,7 @@ from open_range.protocols import SnapshotSpec, TaskSpec
 from open_range.server.models import RangeAction, RangeObservation, RangeState
 
 if TYPE_CHECKING:
+    from open_range.server.compose_runner import BootedSnapshotProject
     from open_range.server.runtime import ManagedSnapshotRuntime
 
 logger = logging.getLogger(__name__)
@@ -123,6 +124,7 @@ class RangeEnvironment(_BASE):  # type: ignore[misc]
         self._docker_available = docker_available
         self._runtime = runtime
         self._episode_recorded = False
+        self._active_project: "BootedSnapshotProject | None" = None
 
         # Execution mode: "auto", "docker", or "subprocess"
         self._execution_mode = execution_mode
@@ -173,6 +175,11 @@ class RangeEnvironment(_BASE):  # type: ignore[misc]
         the bare hostname is returned as a fallback for test compatibility.
         """
         if self._snapshot and self._snapshot.compose:
+            if (
+                self._active_project is not None
+                and host in self._active_project.containers.container_ids
+            ):
+                return self._active_project.containers.container_ids[host]
             services = self._snapshot.compose.get("services", {})
             if host in services:
                 project = self._snapshot.compose.get(
@@ -505,6 +512,52 @@ class RangeEnvironment(_BASE):  # type: ignore[misc]
                 logger.debug("NPC stop error (ignored): %s", exc)
             self._npc_manager = None
 
+    def _teardown_active_project(self) -> None:
+        """Tear down the currently active runtime-backed episode project."""
+        if self._active_project is None:
+            return
+        project = self._active_project
+        self._active_project = None
+        if self._runtime is None:
+            return
+        try:
+            self._runtime.teardown_snapshot_project(project)
+        except Exception as exc:
+            logger.warning(
+                "Failed to tear down active snapshot project %s: %s",
+                project.project_name,
+                exc,
+            )
+
+    def _activate_runtime_snapshot(
+        self,
+        snapshot: SnapshotSpec,
+        *,
+        episode_id: str,
+    ) -> bool:
+        """Boot a clean project for a runtime-backed admitted snapshot.
+
+        Returns True when the snapshot was activated through the managed
+        runtime and no overlay deployment is needed in-process.
+        """
+        if self._runtime is None or not self._snapshot_id:
+            return False
+        if self._execution_mode != "docker":
+            return False
+        if self._get_docker() is None:
+            return False
+
+        project = self._runtime.activate_snapshot_project(
+            snapshot_id=self._snapshot_id,
+            snapshot=snapshot,
+            episode_id=episode_id,
+        )
+        self._active_project = project
+        compose = dict(snapshot.compose)
+        compose["x-project-name"] = project.project_name
+        snapshot.compose = compose
+        return True
+
     def _refresh_npc_traffic_log(self) -> None:
         """Pull latest NPC activity from the manager into the traffic log."""
         if self._npc_manager is not None:
@@ -851,6 +904,8 @@ class RangeEnvironment(_BASE):  # type: ignore[misc]
             Initial RangeObservation with the challenge briefing.
         """
         self._report_episode_result(completed=False)
+        self._stop_npcs()
+        self._teardown_active_project()
 
         # Select snapshot
         self._snapshot = self._select_snapshot(**kwargs)
@@ -880,8 +935,11 @@ class RangeEnvironment(_BASE):  # type: ignore[misc]
         except Exception:
             pass
 
-        # Deploy snapshot artifacts to running containers
-        self._apply_snapshot(self._snapshot)
+        # Runtime-backed episodes boot a fresh project per reset. Manual/mock
+        # snapshots still use direct artifact application.
+        activated = self._activate_runtime_snapshot(self._snapshot, episode_id=eid)
+        if not activated:
+            self._apply_snapshot(self._snapshot)
 
         # Start NPC traffic for this episode
         self._start_npcs(self._snapshot)
@@ -979,6 +1037,7 @@ class RangeEnvironment(_BASE):  # type: ignore[misc]
             self._report_if_done(obs)
             return obs
 
+
         # Route to container
         target = self._resolve_target(action)
         timeout = timeout_s or self._exec_timeout
@@ -1201,6 +1260,7 @@ class RangeEnvironment(_BASE):  # type: ignore[misc]
         """Release resources (Docker client, NPC manager, episode state)."""
         self._report_episode_result(completed=False)
         self._stop_npcs()
+        self._teardown_active_project()
         if self._docker_client is not None:
             try:
                 self._docker_client.close()

src/open_range/server/runtime.py

@@ -1053,6 +1053,57 @@ class ManagedSnapshotRuntime:
         rendered.compose = yaml.safe_load(compose_path.read_text(encoding="utf-8")) or {}
         return rendered
 
+    def activate_snapshot_project(
+        self,
+        *,
+        snapshot_id: str,
+        snapshot: SnapshotSpec,
+        episode_id: str | None = None,
+    ) -> BootedSnapshotProject:
+        """Boot a fresh per-episode project for an admitted snapshot.
+
+        This is the runtime-facing execution path used by RangeEnvironment.
+        It keeps episode state isolated by booting a new compose project from
+        the admitted artifact bundle rather than layering files onto a
+        long-lived shared stack.
+        """
+        self.start()
+
+        materialized = snapshot
+        artifacts_dir = self._artifacts_dir(snapshot_id)
+        if not artifacts_dir.exists():
+            materialized = self._materialize_snapshot(snapshot, snapshot_id)
+
+        project_name_seed = snapshot_id
+        if episode_id:
+            project_name_seed = f"{snapshot_id}-{episode_id}"
+        project_name = self.compose_runner.project_name_for(project_name_seed)
+
+        project: BootedSnapshotProject | None = None
+        try:
+            project = self.compose_runner.boot(
+                snapshot_id=snapshot_id,
+                artifacts_dir=artifacts_dir,
+                compose=materialized.compose,
+                project_name=project_name,
+            )
+            self._apply_rendered_payloads(snapshot_id, project.containers, materialized)
+            return project
+        except Exception:
+            if project is not None:
+                try:
+                    self.compose_runner.teardown(project)
+                except Exception:  # noqa: BLE001
+                    logger.warning(
+                        "Failed to tear down project %s after activation failure",
+                        project.project_name,
+                    )
+            raise
+
+    def teardown_snapshot_project(self, project: BootedSnapshotProject) -> None:
+        """Tear down a previously activated episode project."""
+        self.compose_runner.teardown(project)
+
     def _run_live_admission(self, snapshot: SnapshotSpec, snapshot_id: str) -> None:
         project: BootedSnapshotProject | None = None
         try:

tests/test_runtime.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+from pathlib import Path
+
 import pytest
 
-from open_range.protocols import CheckResult
+from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
 from open_range.server.compose_runner import BootedSnapshotProject
 from open_range.server.environment import RangeEnvironment
 from open_range.server.runtime import ManagedSnapshotRuntime
@@ -243,6 +245,73 @@ class TestManagedSnapshotRuntime:
         finally:
             runtime.stop()
 
+    def test_activate_snapshot_project_uses_unique_episode_project_name(
+        self,
+        tier1_manifest,
+        tmp_path,
+    ):
+        class FakeContainers:
+            def __init__(self) -> None:
+                self.exec_calls: list[tuple[str, str]] = []
+                self.cp_calls: list[tuple[str, str, str]] = []
+
+            async def exec(self, container: str, cmd: str, **kwargs) -> str:
+                self.exec_calls.append((container, cmd))
+                return "ok"
+
+            async def cp(self, container: str, src: str, dest: str) -> None:
+                self.cp_calls.append((container, src, dest))
+
+            async def is_healthy(self, container: str) -> bool:
+                return True
+
+        class FakeComposeRunner:
+            def __init__(self) -> None:
+                self.boot_calls: list[tuple[str, str, str | None]] = []
+                self.teardown_calls: list[str] = []
+                self.containers = FakeContainers()
+
+            def project_name_for(self, snapshot_id: str) -> str:
+                return f"openrange-{snapshot_id}"[:63]
+
+            def boot(self, *, snapshot_id, artifacts_dir, compose, project_name=None):
+                self.boot_calls.append((snapshot_id, str(artifacts_dir), project_name))
+                return BootedSnapshotProject(
+                    project_name=project_name or f"openrange-{snapshot_id}",
+                    compose_file=artifacts_dir / "docker-compose.yml",
+                    artifacts_dir=artifacts_dir,
+                    containers=self.containers,  # type: ignore[arg-type]
+                )
+
+            def teardown(self, project):
+                self.teardown_calls.append(project.project_name)
+
+        compose_runner = FakeComposeRunner()
+        runtime = ManagedSnapshotRuntime(
+            manifest=tier1_manifest,
+            store_dir=tmp_path / "snapshots",
+            pool_size=1,
+            refill_enabled=False,
+            compose_runner=compose_runner,  # type: ignore[arg-type]
+        )
+
+        runtime.start()
+        try:
+            admitted = runtime.acquire_snapshot()
+            project = runtime.activate_snapshot_project(
+                snapshot_id=admitted.snapshot_id,
+                snapshot=admitted.snapshot,
+                episode_id="episode-123",
+            )
+            assert compose_runner.boot_calls
+            _, artifacts_dir, project_name = compose_runner.boot_calls[0]
+            assert artifacts_dir.endswith(f"{admitted.snapshot_id}/artifacts")
+            assert project_name == f"openrange-{admitted.snapshot_id}-episode-123"
+            runtime.teardown_snapshot_project(project)
+            assert compose_runner.teardown_calls == [project.project_name]
+        finally:
+            runtime.stop()
+
 
 class TestEnvironmentRuntimeIntegration:
     def test_reset_uses_managed_runtime_snapshot(self, tier1_manifest, tmp_path):
@@ -299,3 +368,75 @@ class TestEnvironmentRuntimeIntegration:
         finally:
             env.close()
             runtime.stop()
+
+    def test_reset_activates_clean_runtime_project_and_tears_down_previous(self):
+        class FakeRuntime:
+            def __init__(self, snapshot: SnapshotSpec) -> None:
+                self.snapshot = snapshot
+                self.activate_calls: list[tuple[str, str | None]] = []
+                self.teardown_calls: list[str] = []
+                self.recorded: list[bool] = []
+
+            def acquire_snapshot(self):
+                return type(
+                    "Admitted",
+                    (),
+                    {"snapshot_id": "snap-001", "snapshot": self.snapshot},
+                )()
+
+            def get_snapshot(self, snapshot_id: str):
+                assert snapshot_id == "snap-001"
+                return self.acquire_snapshot()
+
+            def activate_snapshot_project(self, *, snapshot_id, snapshot, episode_id=None):
+                self.activate_calls.append((snapshot_id, episode_id))
+                return BootedSnapshotProject(
+                    project_name=f"project-{episode_id}",
+                    compose_file=Path("/tmp/docker-compose.yml"),
+                    artifacts_dir=Path("/tmp"),
+                    containers=ContainerSet(
+                        project_name=f"project-{episode_id}",
+                        container_ids={"web": "cid-web", "attacker": "cid-attacker", "siem": "cid-siem"},
+                    ),
+                )
+
+            def teardown_snapshot_project(self, project):
+                self.teardown_calls.append(project.project_name)
+
+            def record_episode_result(self, **kwargs):
+                self.recorded.append(bool(kwargs.get("completed", False)))
+
+        snapshot = SnapshotSpec(
+            topology={"hosts": ["attacker", "siem", "web"]},
+            compose={"services": {"attacker": {}, "siem": {}, "web": {}}},
+            task={"red_briefing": "Go.", "blue_briefing": "Watch."},
+        )
+        runtime = FakeRuntime(snapshot)
+        env = RangeEnvironment(
+            runtime=runtime,  # type: ignore[arg-type]
+            docker_available=True,
+            execution_mode="docker",
+        )
+
+        env._get_docker = lambda: object()  # type: ignore[method-assign]
+        apply_calls: list[str] = []
+        env._apply_snapshot = lambda snapshot: apply_calls.append("overlay")  # type: ignore[method-assign]
+        env._start_npcs = lambda snapshot: None  # type: ignore[method-assign]
+
+        try:
+            env.reset(episode_id="ep-1")
+            assert runtime.activate_calls == [("snap-001", "ep-1")]
+            assert apply_calls == []
+            assert env.snapshot is not None
+            assert env.snapshot.compose["x-project-name"] == "project-ep-1"
+            assert env._container_name("web") == "cid-web"
+
+            env.reset(episode_id="ep-2")
+            assert runtime.activate_calls == [("snap-001", "ep-1"), ("snap-001", "ep-2")]
+            assert runtime.teardown_calls == ["project-ep-1"]
+            assert env.snapshot is not None
+            assert env.snapshot.compose["x-project-name"] == "project-ep-2"
+        finally:
+            env.close()
+
+        assert runtime.teardown_calls == ["project-ep-1", "project-ep-2"]