Implement issues #72 #74 #75 env grounding and console scope

src/open_range/server/app.py

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

src/open_range/server/console.py

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

src/open_range/server/environment.py

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

tests/test_console_context.py

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

tests/test_environment.py

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