Wire console to real episode state

src/open_range/server/console.py

@@ -6,6 +6,7 @@ the range environment state, viewing action history, and triggering resets.
 
 from __future__ import annotations
 
+import copy
 import time
 from typing import Any
 
@@ -20,6 +21,7 @@ console_router = APIRouter(prefix="/console", tags=["console"])
 
 _action_history: list[dict[str, Any]] = []
 _MAX_HISTORY = 50  # keep more than 20 internally, but serve 20
+_published_episode: dict[str, dict[str, Any]] | None = None
 
 
 def record_action(action_record: dict[str, Any]) -> None:
@@ -39,6 +41,49 @@ def get_history(limit: int = 20) -> list[dict[str, Any]]:
     return list(reversed(_action_history[-limit:]))
 
 
+def publish_episode(snapshot: Any, state: Any) -> None:
+    """Publish the latest episode summary for console readers.
+
+    This is the bridge used by real reset/step traffic from HTTP handlers,
+    where OpenEnv creates short-lived environment instances per request.
+    """
+    global _published_episode
+
+    topo = snapshot.topology if snapshot and isinstance(snapshot.topology, dict) else {}
+    hosts = topo.get("hosts", [])
+    zones = topo.get("zones", {})
+    vuln_count = 0
+    if snapshot is not None and getattr(snapshot, "truth_graph", None) is not None:
+        vuln_count = len(getattr(snapshot.truth_graph, "vulns", []) or [])
+
+    _published_episode = {
+        "snapshot": {
+            "id": getattr(state, "episode_id", None),
+            "tier": topo.get("tier", getattr(state, "tier", 1)),
+            "hosts": list(hosts) if isinstance(hosts, list) else [],
+            "zones": copy.deepcopy(zones) if isinstance(zones, dict) else {},
+            "vuln_count": vuln_count,
+        },
+        "episode": {
+            "step_count": int(getattr(state, "step_count", 0) or 0),
+            "flags_found": len(getattr(state, "flags_found", []) or []),
+            "mode": getattr(state, "mode", ""),
+            "services_status": copy.deepcopy(getattr(state, "services_status", {}) or {}),
+        },
+    }
+
+
+def clear_episode() -> None:
+    """Clear the published episode summary."""
+    global _published_episode
+    _published_episode = None
+
+
+def get_published_episode() -> dict[str, dict[str, Any]] | None:
+    """Return a defensive copy of the published episode summary."""
+    return copy.deepcopy(_published_episode)
+
+
 # ---------------------------------------------------------------------------
 # API routes
 # ---------------------------------------------------------------------------
@@ -48,6 +93,15 @@ def get_history(limit: int = 20) -> list[dict[str, Any]]:
 async def api_snapshot(request: Request) -> JSONResponse:
     """Return current snapshot metadata (no truth graph or flags)."""
     ctx = _get_env_context(request)
+    published = ctx.get("published_episode")
+    if published is not None:
+        return JSONResponse({
+            **published["snapshot"],
+            "state_scope": ctx["state_scope"],
+            "session_id": ctx["session_id"],
+            "warning": ctx["warning"],
+        })
+
     env = ctx["env"]
     snapshot = env.snapshot
     if snapshot is None:
@@ -84,6 +138,15 @@ async def api_snapshot(request: Request) -> JSONResponse:
 async def api_episode(request: Request) -> JSONResponse:
     """Return current episode state."""
     ctx = _get_env_context(request)
+    published = ctx.get("published_episode")
+    if published is not None:
+        return JSONResponse({
+            **published["episode"],
+            "state_scope": ctx["state_scope"],
+            "session_id": ctx["session_id"],
+            "warning": ctx["warning"],
+        })
+
     env = ctx["env"]
     state = env.state
     return JSONResponse({
@@ -120,8 +183,9 @@ def _get_env_context(request: Request) -> dict[str, Any]:
 
     Priority:
     1. Active OpenEnv WebSocket session environment (session-scoped truth)
-    2. ``app.state.env`` fallback environment (global app scope)
-    3. Lazily created fallback environment (tests/dev)
+    2. Published reset/step state from real HTTP traffic
+    3. ``app.state.env`` fallback environment (global app scope)
+    4. Lazily created fallback environment (tests/dev)
     """
     app = request.app
 
@@ -132,6 +196,7 @@ def _get_env_context(request: Request) -> dict[str, Any]:
             session_id, env = next(iter(sessions.items()))
             return {
                 "env": env,
+                "published_episode": None,
                 "state_scope": "websocket_session",
                 "session_id": session_id,
                 "warning": None,
@@ -144,6 +209,7 @@ def _get_env_context(request: Request) -> dict[str, Any]:
         )
         return {
             "env": sessions[selected_id],
+            "published_episode": None,
             "state_scope": "websocket_session",
             "session_id": selected_id,
             "warning": (
@@ -152,9 +218,23 @@ def _get_env_context(request: Request) -> dict[str, Any]:
             ),
         }
 
+    published = get_published_episode()
+    if published is not None:
+        return {
+            "env": None,
+            "published_episode": published,
+            "state_scope": "published_episode",
+            "session_id": None,
+            "warning": (
+                "No active WebSocket session found; console is showing the most "
+                "recent reset/step state observed by the server."
+            ),
+        }
+
     if hasattr(app.state, "env"):
         return {
             "env": app.state.env,
+            "published_episode": None,
             "state_scope": "app_state_env",
             "session_id": None,
             "warning": (
@@ -170,6 +250,7 @@ def _get_env_context(request: Request) -> dict[str, Any]:
         app.state._fallback_env = RangeEnvironment(docker_available=False)
     return {
         "env": app.state._fallback_env,
+        "published_episode": None,
         "state_scope": "fallback_env",
         "session_id": None,
         "warning": "Console is using a fallback environment (no server session available).",
@@ -284,6 +365,7 @@ _CONSOLE_HTML = """\
   }
   .history-item .mode-red { color: var(--red); }
   .history-item .mode-blue { color: var(--accent); }
+  .history-item .mode-system { color: var(--yellow); }
   .history-item .ts {
     color: var(--text-dim);
     font-size: 11px;
@@ -424,7 +506,11 @@ function renderHistory(items) {
     return;
   }
   el.innerHTML = items.map(function(it) {
-    const modeClass = it.mode === "red" ? "mode-red" : "mode-blue";
+    const modeClass = it.mode === "red"
+      ? "mode-red"
+      : it.mode === "blue"
+        ? "mode-blue"
+        : "mode-system";
     return '<div class="history-item">' +
       '<span class="ts">' + fmtTime(it.time) + '</span>' +
       '<span class="step">step ' + (it.step || "-") + '</span> ' +

src/open_range/server/environment.py

@@ -924,6 +924,15 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         except Exception:
             pass
 
+    def _record_console_action(self, mode: str, action_record: dict[str, Any]) -> None:
+        """Record a console-visible action without coupling to console internals."""
+        try:
+            from open_range.server.console import record_action
+
+            record_action({"mode": mode, **action_record})
+        except Exception:
+            pass
+
     # -----------------------------------------------------------------
     # Snapshot selection
     # -----------------------------------------------------------------
@@ -1462,6 +1471,16 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         )
 
         self._publish_console_state()
+        self._record_console_action(
+            "system",
+            {
+                "step": 0,
+                "command": "reset",
+                "cmd_name": "reset",
+                "time": time.time(),
+                "episode_id": eid,
+            },
+        )
         return RangeObservation(stdout=briefing)
 
     def step(
@@ -1520,6 +1539,15 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         }
 
         if cmd_name in meta_handlers:
+            self._record_console_action(
+                action.mode,
+                {
+                    "step": self._state.step_count,
+                    "command": action.command,
+                    "cmd_name": cmd_name,
+                    "time": time.time(),
+                },
+            )
             obs = meta_handlers[cmd_name](action)
             self._refresh_services_status()
             obs = self._apply_rewards(action, obs)
@@ -1550,12 +1578,7 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             self._red_history.append(action_record)
         else:
             self._blue_history.append(action_record)
-        try:
-            from open_range.server.console import record_action
-
-            record_action({"mode": action.mode, **action_record})
-        except Exception:
-            pass
+        self._record_console_action(action.mode, action_record)
 
         # Check for milestone completion (#17)
         milestone = self._check_milestone(stdout)

tests/test_console.py

@@ -1,21 +1,22 @@
-"""Tests for the operator debugging console (issue #28).
+"""Tests for the operator debugging console.
 
 Uses Starlette's TestClient against the OpenEnv app with console router.
 No Docker dependency.
 
-Note: OpenEnv HTTP endpoints are stateless (each creates a new env instance).
-Console API uses a fallback env stored on app.state.  History is recorded
-via the module-level record_action() / clear_history() helpers.
+The console prefers live WebSocket session state, then published reset/step
+state from real traffic, then a local fallback env for dev/test use.
 """
 
 from __future__ import annotations
 
+from unittest.mock import patch
+
 import pytest
 from starlette.testclient import TestClient
 
 from open_range.protocols import SnapshotSpec
 from open_range.server.app import create_app
-from open_range.server.console import clear_history, record_action
+from open_range.server.console import clear_episode, clear_history, record_action
 from open_range.server.environment import RangeEnvironment
 
 _TEST_SNAPSHOT = SnapshotSpec(
@@ -36,8 +37,11 @@ def client():
     # Store a shared env so console API endpoints can access state
     env = RangeEnvironment(docker_available=False)
     app.state.env = env
+    clear_episode()
+    clear_history()
+    yield TestClient(app)
+    clear_episode()
     clear_history()
-    return TestClient(app)
 
 
 @pytest.fixture()
@@ -101,6 +105,27 @@ class TestSnapshotAPI:
         assert "flags" not in data
         assert "golden_path" not in data
 
+    def test_http_reset_publishes_snapshot_to_console(self):
+        with patch(
+            "open_range.server.environment.RangeEnvironment._select_snapshot",
+            return_value=_TEST_SNAPSHOT,
+        ), patch(
+            "open_range.server.environment.RangeEnvironment._ensure_clean_reset_path",
+        ):
+            app = create_app()
+            clear_episode()
+            clear_history()
+            client = TestClient(app)
+            resp = client.post("/reset", json={"episode_id": "http_reset_1"})
+            assert resp.status_code == 200
+
+            data = client.get("/console/api/snapshot").json()
+            assert data["id"] == "http_reset_1"
+            assert data["state_scope"] == "published_episode"
+            assert data["hosts"] == ["attacker", "siem"]
+            clear_episode()
+            clear_history()
+
 
 # ===================================================================
 # GET /console/api/episode
@@ -133,6 +158,25 @@ class TestEpisodeAPI:
         data = client.get("/console/api/episode").json()
         assert data["step_count"] == 1
 
+    def test_http_reset_publishes_episode_to_console(self):
+        with patch(
+            "open_range.server.environment.RangeEnvironment._select_snapshot",
+            return_value=_TEST_SNAPSHOT,
+        ), patch(
+            "open_range.server.environment.RangeEnvironment._ensure_clean_reset_path",
+        ):
+            app = create_app()
+            clear_episode()
+            clear_history()
+            client = TestClient(app)
+            client.post("/reset", json={"episode_id": "http_reset_2"})
+            data = client.get("/console/api/episode").json()
+            assert data["step_count"] == 0
+            assert data["mode"] == "red"
+            assert data["state_scope"] == "published_episode"
+            clear_episode()
+            clear_history()
+
 
 # ===================================================================
 # GET /console/api/history
@@ -176,9 +220,32 @@ class TestHistoryAPI:
         env.reset(snapshot=_TEST_SNAPSHOT)
         env.step(RangeAction(command="nmap -sV web", mode="red"))
         data = client.get("/console/api/history").json()
-        assert len(data) == 1
+        assert len(data) == 2
         assert data[0]["command"] == "nmap -sV web"
         assert data[0]["mode"] == "red"
+        assert data[1]["command"] == "reset"
+        assert data[1]["mode"] == "system"
+
+    def test_history_records_meta_step_commands(self, client: TestClient, env: RangeEnvironment):
+        from open_range.server.models import RangeAction
+
+        env.reset(snapshot=_TEST_SNAPSHOT)
+        env.step(RangeAction(command="submit_finding suspicious scan on web", mode="blue"))
+        data = client.get("/console/api/history").json()
+        assert data[0]["command"] == "submit_finding suspicious scan on web"
+        assert data[0]["mode"] == "blue"
+
+    def test_history_reset_clears_prior_entries_and_records_reset(self, client: TestClient, env: RangeEnvironment):
+        import time
+
+        record_action({"step": 99, "command": "old", "mode": "red", "time": time.time()})
+
+        env.reset(snapshot=_TEST_SNAPSHOT, episode_id="history_reset")
+        data = client.get("/console/api/history").json()
+        assert len(data) == 1
+        assert data[0]["command"] == "reset"
+        assert data[0]["mode"] == "system"
+        assert data[0]["episode_id"] == "history_reset"
 
     def test_history_max_20(self, client: TestClient):
         """History API should return at most 20 entries."""
@@ -188,3 +255,22 @@ class TestHistoryAPI:
             record_action({"step": i, "command": f"cmd_{i}", "mode": "red", "time": time.time()})
         data = client.get("/console/api/history").json()
         assert len(data) == 20
+
+    def test_http_reset_records_history(self):
+        with patch(
+            "open_range.server.environment.RangeEnvironment._select_snapshot",
+            return_value=_TEST_SNAPSHOT,
+        ), patch(
+            "open_range.server.environment.RangeEnvironment._ensure_clean_reset_path",
+        ):
+            app = create_app()
+            clear_episode()
+            clear_history()
+            client = TestClient(app)
+            client.post("/reset", json={"episode_id": "http_reset_3"})
+            data = client.get("/console/api/history").json()
+            assert len(data) == 1
+            assert data[0]["command"] == "reset"
+            assert data[0]["mode"] == "system"
+            clear_episode()
+            clear_history()

tests/test_console_bridge.py

@@ -0,0 +1,100 @@
+"""Focused console bridge tests without TestClient transport."""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from open_range.protocols import SnapshotSpec
+from open_range.server.app import create_app
+from open_range.server.console import (
+    api_episode,
+    api_history,
+    api_snapshot,
+    clear_episode,
+    clear_history,
+    get_history,
+)
+from open_range.server.environment import RangeEnvironment
+from open_range.server.models import RangeAction
+
+_TEST_SNAPSHOT = SnapshotSpec(
+    topology={"hosts": ["attacker", "siem"]},
+    flags=[],
+    golden_path=[],
+    task={
+        "red_briefing": "Console bridge test mode.",
+        "blue_briefing": "Console bridge test mode.",
+    },
+)
+
+
+def _request(app):
+    return SimpleNamespace(app=app)
+
+
+def _json_response_payload(response) -> dict | list:
+    return json.loads(response.body.decode())
+
+
+def test_http_reset_publishes_console_snapshot_and_episode():
+    clear_episode()
+    clear_history()
+    with patch(
+        "open_range.server.environment.RangeEnvironment._select_snapshot",
+        return_value=_TEST_SNAPSHOT,
+    ), patch(
+        "open_range.server.environment.RangeEnvironment._ensure_clean_reset_path",
+    ):
+        app = create_app()
+        env = app.state.openenv_server._env_factory()
+        try:
+            env.reset(episode_id="http_console_ep")
+        finally:
+            env.close()
+
+        snapshot = _json_response_payload(_run(api_snapshot(_request(app))))
+        episode = _json_response_payload(_run(api_episode(_request(app))))
+
+    assert snapshot["id"] == "http_console_ep"
+    assert snapshot["hosts"] == ["attacker", "siem"]
+    assert snapshot["state_scope"] == "published_episode"
+    assert episode["step_count"] == 0
+    assert episode["mode"] == "red"
+    assert episode["state_scope"] == "published_episode"
+
+
+def test_environment_reset_clears_history_and_records_reset():
+    clear_episode()
+    clear_history()
+    env = RangeEnvironment(docker_available=False)
+
+    env.reset(snapshot=_TEST_SNAPSHOT, episode_id="console_reset_ep")
+
+    history = get_history()
+    assert len(history) == 1
+    assert history[0]["command"] == "reset"
+    assert history[0]["mode"] == "system"
+    assert history[0]["episode_id"] == "console_reset_ep"
+
+
+def test_environment_meta_steps_record_console_history():
+    clear_episode()
+    clear_history()
+    env = RangeEnvironment(docker_available=False)
+    env.reset(snapshot=_TEST_SNAPSHOT, episode_id="console_meta_ep")
+
+    env.step(RangeAction(command="submit_finding suspicious scan on web", mode="blue"))
+
+    history = _json_response_payload(_run(api_history()))
+    assert len(history) == 2
+    assert history[0]["command"] == "submit_finding suspicious scan on web"
+    assert history[0]["mode"] == "blue"
+    assert history[1]["command"] == "reset"
+
+
+def _run(awaitable):
+    import asyncio
+
+    return asyncio.run(awaitable)

tests/test_console_context.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 
 from types import SimpleNamespace
 
-from open_range.server.console import _get_env_context
+from open_range.protocols import SnapshotSpec
+from open_range.server.console import _get_env_context, clear_episode, publish_episode
 from open_range.server.environment import RangeEnvironment
+from open_range.server.models import RangeState
 
 
 class _Req:
@@ -18,6 +20,7 @@ def _app_with_state(**kwargs):
 
 
 def test_prefers_active_websocket_session_env():
+    clear_episode()
     fallback_env = RangeEnvironment(docker_available=False)
     ws_env = RangeEnvironment(docker_available=False)
     server = SimpleNamespace(
@@ -34,6 +37,7 @@ def test_prefers_active_websocket_session_env():
 
 
 def test_uses_app_state_env_when_no_active_session():
+    clear_episode()
     fallback_env = RangeEnvironment(docker_available=False)
     server = SimpleNamespace(_sessions={}, _session_info={})
     request = _Req(_app_with_state(env=fallback_env, openenv_server=server))
@@ -46,6 +50,7 @@ def test_uses_app_state_env_when_no_active_session():
 
 
 def test_multiple_sessions_selects_most_recent_and_warns():
+    clear_episode()
     older_env = RangeEnvironment(docker_available=False)
     newer_env = RangeEnvironment(docker_available=False)
     server = SimpleNamespace(
@@ -62,3 +67,29 @@ def test_multiple_sessions_selects_most_recent_and_warns():
     assert ctx["state_scope"] == "websocket_session"
     assert ctx["session_id"] == "new"
     assert "active sessions" in (ctx["warning"] or "").lower()
+
+
+def test_uses_published_episode_before_app_state_fallback():
+    clear_episode()
+    snapshot = SnapshotSpec(
+        topology={"hosts": ["attacker"], "zones": {"dmz": ["web"]}, "tier": 2},
+        flags=[],
+        golden_path=[],
+        task={"red_briefing": "r", "blue_briefing": "b"},
+    )
+    publish_episode(
+        snapshot,
+        RangeState(episode_id="published_ep", step_count=4, mode="blue", tier=2),
+    )
+
+    fallback_env = RangeEnvironment(docker_available=False)
+    server = SimpleNamespace(_sessions={}, _session_info={})
+    request = _Req(_app_with_state(env=fallback_env, openenv_server=server))
+
+    ctx = _get_env_context(request)
+    assert ctx["env"] is None
+    assert ctx["published_episode"]["snapshot"]["id"] == "published_ep"
+    assert ctx["state_scope"] == "published_episode"
+    assert ctx["session_id"] is None
+    assert "most recent reset/step state" in (ctx["warning"] or "")
+    clear_episode()