Remove hardcoded fallbacks, add snapshot-driven service lifecycle

- Remove _LEGACY_STOP_DAEMONS, _start_services_legacy, _declared_service_daemons
- _stop_services and _capture_service_pids derive daemons from snapshot.services
- _start_snapshot_services skips when no service specs (no legacy fallback)
- _resolve_target reads topology roles/zones, raises if missing
- _select_snapshot raises RuntimeError when no snapshot provided
- _get_pending_alerts returns [] instead of synthetic fallback
- Renderer auto-populates snapshot.services via generate_service_specs
- start.sh supports ServiceSpec-driven startup alongside legacy path
- Tests updated: explicit snapshot in all reset() calls, 114 env tests pass
- 70 new service spec tests (models, generation, lifecycle, renderer)

README.md

@@ -161,7 +161,7 @@ Difficulty grows horizontally — more hosts, zones, and chained attack surface.
 | GET | `/state` | Current episode state |
 | WS | `/ws` | WebSocket session |
 
-Compatible with `openenv` when installed; standalone FastAPI fallback otherwise.
+Built directly on the OpenEnv HTTP/WebSocket contract.
 
 ## Docs
 

docs/openenv-compliance.md

@@ -69,18 +69,17 @@ flowchart TD
 
 1. **Don't redeclare `done` or `reward` on Observation.** The base class already has them. `RangeObservation` correctly inherits them.
 2. **Don't redeclare `episode_id` or `step_count` on State.** The base class already has them. `RangeState` correctly inherits them.
-3. **Pass the CLASS to `create_app()`, not an instance.** Each WebSocket session gets its own instance. The standalone fallback also creates per-session instances for WebSocket.
-4. **Action uses `extra="forbid"` (via openenv base).** Unknown fields cause validation errors. Keep actions minimal. Note: the fallback `Action` stub does not enforce `extra="forbid"`.
+3. **Pass the CLASS or factory to `create_app()`, not an instance.** Each WebSocket session gets its own instance.
+4. **Action uses `extra="forbid"` (via openenv base).** Unknown fields cause validation errors. Keep actions minimal.
 5. **State uses `extra="allow"`.** You can add any fields you want.
 6. **`reset()` returns ObsT (server-side), `StepResult[ObsT]` (client-side).** The server wraps it.
-7. **All openenv imports are guarded with try/except.** Models, environment, client, and app all fall back gracefully when openenv is not installed.
+7. **Shared models live outside `server/`.** Clients import `open_range.models`, not `open_range.server.*`.
 
 ## API Signatures (Exact)
 
 ```python
 # Server-side (src/open_range/server/environment.py)
 class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
-    # Falls back to object base when openenv is not installed
     SUPPORTS_CONCURRENT_SESSIONS = False
 
     def __init__(self, max_steps: int = 100, exec_timeout: float = 30.0,
@@ -94,15 +93,12 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
 
 # Client-side (src/open_range/client/client.py)
 class OpenRangeEnv(EnvClient[RangeAction, RangeObservation, RangeState]):
-    # Falls back to a stub class when openenv is not installed
     def _step_payload(self, action: RangeAction) -> dict: ...
     def _parse_result(self, payload: dict) -> StepResult[RangeObservation]: ...
     def _parse_state(self, payload: dict) -> RangeState: ...
 
 # App factory (src/open_range/server/app.py)
-# Tries openenv's create_app first:
 app = create_app(RangeEnvironment, RangeAction, RangeObservation, env_name="open_range")
-# Falls back to standalone FastAPI app with equivalent HTTP + WebSocket endpoints
 
 # Entry point (src/open_range/server/__main__.py)
 # python -m open_range.server [--host HOST] [--port PORT] [--reload] [--log-level LEVEL]

docs/red-blue-agents.md

@@ -113,7 +113,7 @@ The **orchestration layer** (not the agent) controls `reset()` and episode lifec
 
 ```python
 from open_range.agents.protocol import EpisodeMetrics, EpisodeResult
-from open_range.server.models import RangeAction
+from open_range.models import RangeAction
 
 def run_episode(
     env: object,

examples/demo.py

@@ -30,8 +30,8 @@ from open_range.protocols import (
     TruthGraph,
     Vulnerability,
 )
+from open_range.models import RangeAction, RangeObservation
 from open_range.server.environment import RangeEnvironment
-from open_range.server.models import RangeAction, RangeObservation
 from open_range.training.trajectory import TrajectoryLogger
 
 

manifests/tier1_basic.yaml

@@ -424,7 +424,7 @@ credential_policy:
   shared_accounts:
     - account: admin
       host: web
-      description: "Shared by IT team (dthompson, rkim) for emergency access to the web server"
+      description: "Shared by IT team (dthompson, rchen) for emergency access to the web server"
   service_accounts:
     - account: svc_backup
       host: db
@@ -480,27 +480,27 @@ trust_relationships:
       Derek set up Karen's workstation and knows her login credentials.
       He occasionally logs in as Karen to troubleshoot billing portal issues.
 
-  - from: schen
-    to: lpark
+  - from: apatel
+    to: bmorris
     type: reports_to
     description: >-
-      Lisa (office manager) handles Sarah's (medical director) calendar and
-      email when Sarah is in clinic.  Lisa has delegated access to Sarah's
+      Anita (office manager) handles Brian's (CEO) calendar and
+      email when Brian is traveling.  Anita has delegated access to Brian's
       mailbox.
 
-  - from: jrodriguez
+  - from: ldunn
     to: dthompson
     type: trusts_email
     description: >-
-      Julia (compliance) always asks Derek for access changes via email and
-      never verifies requests by phone.  A spoofed email from Julia would
+      Linda (compliance officer) always asks Derek for access changes via email
+      and never verifies requests by phone.  A spoofed email from Linda would
       likely be actioned without question.
 
-  - from: rkim
+  - from: rchen
     to: dthompson
     type: shares_credentials
     description: >-
-      Ryan (security contractor) and Derek share the 'admin' account on the
+      Rachel (security contractor) and Derek share the 'admin' account on the
       web server.  Neither can distinguish the other's sessions in logs.
 
 # ---------------------------------------------------------------------------

src/open_range/__init__.py

@@ -1,12 +1,8 @@
 """OpenRange public package surface."""
 
 from open_range.client.client import OpenRangeEnv
+from open_range.models import RangeAction, RangeObservation, RangeState
 from open_range.server.environment import RangeEnvironment
-from open_range.server.models import (
-    RangeAction,
-    RangeObservation,
-    RangeState,
-)
 
 __all__ = [
     "OpenRangeEnv",

src/open_range/agents/episode.py

@@ -102,7 +102,7 @@ def run_episode(
     Returns:
         ``EpisodeResult`` with trajectories, metrics, and outcome.
     """
-    from open_range.server.models import RangeAction
+    from open_range.models import RangeAction
 
     # Reset environment
     obs = env.reset()

src/open_range/builder/builder.py

@@ -504,7 +504,10 @@ def _parse_llm_response(raw_json: str) -> SnapshotSpec:
     elif isinstance(evidence_raw, list):
         for item in evidence_raw:
             if isinstance(item, dict):
-                evidence_spec.append(EvidenceItem(**item))
+                try:
+                    evidence_spec.append(EvidenceItem(**item))
+                except Exception:  # noqa: BLE001
+                    logger.warning("Skipping malformed evidence item: %s", item)
 
     # Map NPC personas
     npc_personas = []
@@ -1192,7 +1195,7 @@ def render_template_payloads(
                 (
                     "USE flags;\n"
                     "INSERT INTO secrets(flag_name, flag) "
-                    f"VALUES ('{flag.id}', '{flag.value}');\n"
+                    f"VALUES ('{_sql_escape(flag.id)}', '{_sql_escape(flag.value)}');\n"
                 ),
             )
             if vuln_types.intersection({"weak_creds", "idor"}):
@@ -1242,6 +1245,15 @@ def _append_sql(existing: str, fragment: str) -> str:
     return f"{existing.rstrip()}\n{fragment}"
 
 
+def _sql_escape(value: str) -> str:
+    """Escape a string for use in a SQL single-quoted literal.
+
+    Replaces single quotes with doubled single quotes and backslashes
+    with doubled backslashes to prevent SQL injection in static SQL files.
+    """
+    return value.replace("\\", "\\\\").replace("'", "''")
+
+
 def _predictable_user_password(
     username: str,
     *,

src/open_range/builder/npc/actions.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import logging
 import re
+import shlex
 import time
 from typing import Any
 
@@ -17,12 +18,53 @@ from open_range.protocols import ContainerSet, NPCAction, NPCPersona, SnapshotSp
 logger = logging.getLogger(__name__)
 
 
+# ---------------------------------------------------------------------------
+# Host resolution -- resolve logical roles to actual topology hostnames
+# ---------------------------------------------------------------------------
+
+
+def _resolve_host(
+    snapshot: SnapshotSpec,
+    keywords: list[str],
+    fallback: str,
+) -> str:
+    """Resolve a logical role to an actual hostname from the snapshot topology.
+
+    Searches ``snapshot.topology["hosts"]`` for a host whose name or services
+    match any of the given *keywords*.  Returns the first match, or *fallback*
+    if the topology is empty or no match is found.
+
+    This mirrors the keyword-matching pattern used in ``npc_manager.py``
+    (``_host_matches_keywords`` / ``_ROLE_SERVICE_KEYWORDS``).
+    """
+    hosts = snapshot.topology.get("hosts") or []
+    for host in hosts:
+        if isinstance(host, str):
+            # Plain string host name -- match against keywords directly
+            host_lower = host.lower()
+            for kw in keywords:
+                if kw.lower() in host_lower:
+                    return host
+            continue
+        if not isinstance(host, dict):
+            continue
+        host_name = (host.get("name") or "").lower()
+        services = [s.lower() for s in (host.get("services") or [])]
+        for kw in keywords:
+            kw_lower = kw.lower()
+            if kw_lower in host_name or any(kw_lower in svc for svc in services):
+                return host.get("name", fallback)
+    return fallback
+
+
 class NPCActionExecutor:
     """Execute NPC actions inside Docker containers.
 
     At init, extracts available pages, shares, DB tables, users, and
     credentials from the snapshot so every action targets real resources
-    in this environment.
+    in this environment.  Container names are resolved from the snapshot
+    topology via keyword matching, so the executor works with any host
+    naming convention (not just the default tier-1 names).
     """
 
     def __init__(self, containers: ContainerSet, snapshot: SnapshotSpec) -> None:
@@ -36,6 +78,13 @@ class NPCActionExecutor:
         self._db_creds = _extract_db_credentials(snapshot)
         self._ssh_creds = _extract_ssh_credentials(snapshot)
 
+        # Resolve logical roles to actual hostnames from the topology
+        self._host_web = _resolve_host(snapshot, ["nginx", "apache", "httpd", "web", "php-fpm"], "web")
+        self._host_mail = _resolve_host(snapshot, ["postfix", "sendmail", "dovecot", "mail"], "mail")
+        self._host_db = _resolve_host(snapshot, ["mysql", "mariadb", "postgres", "mongodb"], "db")
+        self._host_siem = _resolve_host(snapshot, ["rsyslog", "elasticsearch", "siem", "splunk"], "siem")
+        self._host_files = _resolve_host(snapshot, ["samba", "smb", "files", "nfs"], "files")
+
     # ------------------------------------------------------------------
     # Routine actions (autonomous workday)
     # ------------------------------------------------------------------
@@ -70,9 +119,11 @@ class NPCActionExecutor:
         if path == "/" and self._pages:
             import random
             path = random.choice(self._pages)
+        safe_path = shlex.quote(f"http://localhost{path}")
+        safe_ua = shlex.quote(f"Mozilla/5.0 ({username})")
         await self.containers.exec(
-            "web",
-            f'curl -s -o /dev/null -A "Mozilla/5.0 ({username})" "http://localhost{path}"',
+            self._host_web,
+            f"curl -s -o /dev/null -A {safe_ua} {safe_path}",
         )
         return _log(persona, "browse", detail or f"Browsed {path}", f"web:{path}")
 
@@ -92,10 +143,12 @@ class NPCActionExecutor:
             f"To: {recipient}@{self._domain}\\n"
             f"Subject: {detail or 'Update'}\\n\\n{content}"
         )
+        safe_user = shlex.quote(username)
+        safe_msg = shlex.quote(msg)
         await self.containers.exec(
-            "mail",
-            f"mkdir -p /var/mail/{username} "
-            f"&& echo '{msg}' > /var/mail/{username}/sent_{ts_i}.eml",
+            self._host_mail,
+            f"mkdir -p /var/mail/{safe_user} "
+            f"&& echo {safe_msg} > /var/mail/{safe_user}/sent_{ts_i}.eml",
         )
         return _log(persona, "send_email", detail or f"Emailed {recipient}", f"mail:{username}")
 
@@ -112,9 +165,11 @@ class NPCActionExecutor:
         else:
             page = f"/?q={target or 'data'}"
 
+        safe_url = shlex.quote(f"http://localhost{page}")
+        safe_ua = shlex.quote(f"Mozilla/5.0 ({username})")
         await self.containers.exec(
-            "web",
-            f'curl -s -o /dev/null -A "Mozilla/5.0 ({username})" "http://localhost{page}"',
+            self._host_web,
+            f"curl -s -o /dev/null -A {safe_ua} {safe_url}",
         )
         return _log(persona, "lookup", detail or f"Searched: {target}", f"web:{page}")
 
@@ -122,9 +177,10 @@ class NPCActionExecutor:
         """Access a file share that exists in this snapshot."""
         import random
         share = target or (random.choice(self._shares) if self._shares else "general")
+        safe_share = shlex.quote(f"/srv/shares/{share}/")
         await self.containers.exec(
-            "files",
-            f"ls /srv/shares/{share}/ 2>/dev/null || true",
+            self._host_files,
+            f"ls {safe_share} 2>/dev/null || true",
         )
         return _log(persona, "access_share", detail or f"Browsed {share} share", f"files:{share}")
 
@@ -133,11 +189,12 @@ class NPCActionExecutor:
         # Find the login page from snapshot
         login_pages = [p for p in self._pages if "login" in p or "index" in p]
         page = login_pages[0] if login_pages else "/"
+        safe_ua = shlex.quote(f"Mozilla/5.0 ({username})")
+        safe_data = shlex.quote(f"username={username}&password=placeholder")
+        safe_url = shlex.quote(f"http://localhost{page}")
         await self.containers.exec(
-            "web",
-            f'curl -s -o /dev/null -A "Mozilla/5.0 ({username})" '
-            f'-d "username={username}&password=placeholder" '
-            f'"http://localhost{page}"',
+            self._host_web,
+            f"curl -s -o /dev/null -A {safe_ua} -d {safe_data} {safe_url}",
         )
         return _log(persona, "login", detail or "Portal login", "web:access_log")
 
@@ -150,10 +207,16 @@ class NPCActionExecutor:
         else:
             query = "SHOW TABLES"
         db_user, db_pass = self._db_creds
-        cred_flag = f"-u {db_user} -p'{db_pass}'" if db_pass else f"-u {db_user}"
+        safe_user = shlex.quote(db_user)
+        safe_query = shlex.quote(query)
+        if db_pass:
+            safe_pass = shlex.quote(db_pass)
+            cred_flag = f"-u {safe_user} -p{safe_pass}"
+        else:
+            cred_flag = f"-u {safe_user}"
         await self.containers.exec(
-            "db",
-            f'mysql {cred_flag} -e "{query}" 2>/dev/null || true',
+            self._host_db,
+            f"mysql {cred_flag} -e {safe_query} 2>/dev/null || true",
         )
         return _log(persona, "query_db", detail or f"Queried {target or 'database'}", "db:query_log")
 
@@ -185,9 +248,11 @@ class NPCActionExecutor:
                 url = urls[0].rstrip(".")
                 break
         username = _username_from_persona(persona)
+        safe_ua = shlex.quote(f"Mozilla/5.0 ({username})")
+        safe_url = shlex.quote(url)
         await self.containers.exec(
-            "web",
-            f'curl -s -o /dev/null -A "Mozilla/5.0 ({username})" "{url}"',
+            self._host_web,
+            f"curl -s -o /dev/null -A {safe_ua} {safe_url}",
         )
         return _se_log(persona, "click_link", f"Clicked: {url}", "web:access_log", result="success")
 
@@ -195,11 +260,13 @@ class NPCActionExecutor:
         username = _username_from_persona(persona)
         ts_i = int(time.time())
         body = (action.response_content or "acknowledged")[:500]
+        safe_user = shlex.quote(username)
+        safe_msg = shlex.quote(f"From: {username}@{self._domain}\\nSubject: Re\\n\\n{body}")
         await self.containers.exec(
-            "mail",
-            f"mkdir -p /var/mail/{username} "
-            f"&& echo 'From: {username}@{self._domain}\\nSubject: Re\\n\\n{body}' "
-            f"> /var/mail/{username}/sent_{ts_i}.eml",
+            self._host_mail,
+            f"mkdir -p /var/mail/{safe_user} "
+            f"&& echo {safe_msg} "
+            f"> /var/mail/{safe_user}/sent_{ts_i}.eml",
         )
         return _se_log(persona, action.action, "Replied to message", "mail:spool", result="success")
 
@@ -208,26 +275,31 @@ class NPCActionExecutor:
         content = action.response_content or f"username: {username}"
         ts_i = int(time.time())
         # Leaked creds file
-        await self.containers.exec("web", f"echo '{content}' >> /tmp/leaked_{ts_i}.txt")
+        safe_content = shlex.quote(content)
+        await self.containers.exec(self._host_web, f"echo {safe_content} >> /tmp/leaked_{ts_i}.txt")
         # Suspicious login
+        safe_data = shlex.quote(f"username={username}&password=leaked")
         await self.containers.exec(
-            "web",
-            f'curl -s -o /dev/null -A "Mozilla/5.0 (external)" '
-            f'-d "username={username}&password=leaked" "http://localhost/"',
+            self._host_web,
+            f"curl -s -o /dev/null -A {shlex.quote('Mozilla/5.0 (external)')} "
+            f"-d {safe_data} {shlex.quote('http://localhost/')}",
         )
         # SIEM alert
+        safe_name = shlex.quote(persona.name)
         await self.containers.exec(
-            "siem",
-            f'echo "[$(date)] CRED-LEAK: {persona.name} shared credentials" '
+            self._host_siem,
+            f"printf '[%s] CRED-LEAK: %s shared credentials\\n' \"$(date)\" {safe_name} "
             f">> /var/log/siem/consolidated/all.log",
         )
         return _se_log(persona, "share_credentials", f"{persona.name} leaked credentials", "web+siem", result="success")
 
     async def _react_report(self, persona: NPCPersona, action: NPCAction) -> dict[str, Any]:
         detail = "; ".join(action.side_effects) if action.side_effects else "suspicious activity"
+        safe_name = shlex.quote(persona.name)
+        safe_detail = shlex.quote(detail)
         await self.containers.exec(
-            "siem",
-            f'echo "[$(date)] NPC-REPORT: {persona.name}: {detail}" '
+            self._host_siem,
+            f"printf '[%s] NPC-REPORT: %s: %s\\n' \"$(date)\" {safe_name} {safe_detail} "
             f">> /var/log/siem/consolidated/all.log",
         )
         return _se_log(persona, "report_to_IT", detail, "siem:alert", result="blocked")

src/open_range/builder/npc/npc_agent.py

@@ -14,6 +14,8 @@ import json
 import logging
 import os
 import random
+import re
+import shlex
 import time
 from typing import Any
 
@@ -204,6 +206,9 @@ class LLMNPCAgent:
             if "@" in email_acct
             else persona.name.lower().split()[0]
         )
+        # Sanitize mail_user to prevent path traversal / injection
+        if not re.match(r"^[a-zA-Z0-9._-]+$", mail_user):
+            mail_user = re.sub(r"[^a-zA-Z0-9._-]", "_", mail_user)
 
         base_interval = persona.routine.get("action_interval_min", 2)
         interval_s = base_interval * 60
@@ -232,14 +237,16 @@ class LLMNPCAgent:
                 # Red may send real phishing emails via SMTP. Check multiple
                 # mail spool locations for new messages.
                 try:
+                    safe_mail_user = shlex.quote(mail_user)
+                    mail_host = executor._host_mail
                     mail_output = await containers.exec(
-                        "mail",
+                        mail_host,
                         f"{{ find /var/spool/mail/ /var/mail/ "
-                        f"/home/{mail_user}/Maildir/new/ "
-                        f"-newer /tmp/.npc_check_{mail_user} "
+                        f"/home/{safe_mail_user}/Maildir/new/ "
+                        f"-newer /tmp/.npc_check_{safe_mail_user} "
                         f"-type f 2>/dev/null || true; }} | head -3",
                     )
-                    await containers.exec("mail", f"touch /tmp/.npc_check_{mail_user}")
+                    await containers.exec(mail_host, f"touch /tmp/.npc_check_{safe_mail_user}")
 
                     if mail_output and mail_output.strip():
                         for email_file in mail_output.strip().split("\n")[:3]:
@@ -247,7 +254,7 @@ class LLMNPCAgent:
                             if not email_file:
                                 continue
                             content = await containers.exec(
-                                "mail", f"head -50 '{email_file}' 2>/dev/null || true",
+                                mail_host, f"head -50 {shlex.quote(email_file)} 2>/dev/null || true",
                             )
                             if not content or not content.strip():
                                 continue

src/open_range/builder/renderer.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import json
 import logging
+import shlex
 from pathlib import Path
 from pathlib import PurePosixPath
 from typing import Any
@@ -52,6 +53,7 @@ class SnapshotRenderer:
             keep_trailing_newline=True,
             undefined=jinja2.Undefined,
         )
+        self.env.filters["shell_quote"] = shlex.quote
 
     def render(self, spec: SnapshotSpec, output_dir: Path) -> Path:
         """Render all templates and write artifacts to *output_dir*.
@@ -219,12 +221,16 @@ def _build_context(spec: SnapshotSpec) -> dict[str, Any]:
         "db_user": db_user,
         "db_pass": db_pass,
         "db_name": topology.get("db_name", "app_db"),
+        # db_password duplicates db_pass: Dockerfile.db.j2 uses db_pass,
+        # docker-compose.yml.j2 uses db_password.  Keep both for compat.
         "db_password": db_pass,
         "mysql_root_password": topology.get("mysql_root_password", _find_mysql_root_pass(users)),
         "domain": topology.get("domain", "corp.local"),
         "org_name": topology.get("org_name", "Corp"),
         "ldap_admin_pass": topology.get("ldap_admin_pass", "LdapAdm1n!"),
         "smb_shares": _find_smb_shares(spec),
+        "smb_user": _find_smb_user(users),
+        "smb_password": _find_smb_pass(users),
         # Dockerfile.web.j2
         "users": users,
         "app_files": app_files,
@@ -342,6 +348,24 @@ def _find_mysql_root_pass(users: list[dict[str, Any]]) -> str:
     return "r00tP@ss!"
 
 
+def _find_smb_user(users: list[dict[str, Any]]) -> str:
+    """Find the SMB/Samba user from topology users, default to smbuser."""
+    for u in users:
+        hosts = u.get("hosts", [])
+        if "files" in hosts and "admins" not in u.get("groups", []):
+            return u.get("username", "smbuser")
+    return "smbuser"
+
+
+def _find_smb_pass(users: list[dict[str, Any]]) -> str:
+    """Find the SMB/Samba user password."""
+    for u in users:
+        hosts = u.get("hosts", [])
+        if "files" in hosts and "admins" not in u.get("groups", []):
+            return u.get("password", "smbP@ss!")
+    return "smbP@ss!"
+
+
 def _find_smb_shares(spec: SnapshotSpec) -> list[str]:
     """Extract Samba share names from snapshot files dict."""
     shares: set[str] = set()

src/open_range/builder/templates/Dockerfile.web.j2

@@ -20,8 +20,8 @@ RUN mkdir /var/run/sshd && \
 
 # Create app users
 {% for user in users %}
-RUN useradd -m -s /bin/bash {{ user.username }} && \
-    echo '{{ user.username }}:{{ user.password }}' | chpasswd
+RUN useradd -m -s /bin/bash {{ user.username | shell_quote }} && \
+    echo {{ (user.username ~ ':' ~ user.password) | shell_quote }} | chpasswd
 {% endfor %}
 
 # Copy nginx config
@@ -33,8 +33,8 @@ RUN mkdir -p /var/www/portal/admin /var/www/portal/api /var/www/portal/reports
 # Create flag files (if any are on this host)
 {% for flag in flags %}
 {% if flag.host == 'web' and '/' in flag.path %}
-RUN mkdir -p $(dirname {{ flag.path }}) && \
-    echo '{{ flag.value }}' > {{ flag.path }}
+RUN mkdir -p $(dirname {{ flag.path | shell_quote }}) && \
+    echo {{ flag.value | shell_quote }} > {{ flag.path | shell_quote }}
 {% endif %}
 {% endfor %}
 

src/open_range/cli.py

@@ -615,8 +615,8 @@ def episode(
         openrange episode -s snapshots/spec.json --golden-path
         openrange episode -s snapshots/spec.json --interactive --mode both
     """
+    from open_range.models import RangeAction
     from open_range.server.environment import RangeEnvironment
-    from open_range.server.models import RangeAction
 
     spec = _load_snapshot(snapshot)
 

src/open_range/client/client.py

@@ -1,46 +1,51 @@
-"""Typed OpenEnv client for OpenRange.
-
-Falls back to lightweight stubs if openenv is not installed.
-"""
+"""Typed OpenEnv client for OpenRange."""
 
 from __future__ import annotations
 
-from typing import Any, Generic, TypeVar
+from typing import Any
+
+from openenv.core.client_types import StepResult
+from openenv.core.env_client import EnvClient
 
-try:
-    from openenv.core.client_types import StepResult
-    from openenv.core.env_client import EnvClient
-except ImportError:
-    from dataclasses import dataclass, field
+from open_range.models import RangeAction, RangeObservation, RangeState
 
-    _A = TypeVar("_A")
-    _O = TypeVar("_O")
-    _S = TypeVar("_S")
 
-    @dataclass
-    class StepResult(Generic[_O]):  # type: ignore[no-redef]
-        """Minimal stub matching openenv.core.client_types.StepResult."""
+class _SyncOpenRangeEnv:
+    """Synchronous wrapper matching the documented OpenEnv .sync() pattern."""
 
-        observation: Any = None
-        reward: float | int | None = None
-        done: bool = False
-        metadata: dict[str, Any] = field(default_factory=dict)
+    def __init__(self, client: "OpenRangeEnv") -> None:
+        self._client = client
+
+    def __enter__(self) -> "_SyncOpenRangeEnv":
+        return self
 
-    class EnvClient(Generic[_A, _O, _S]):  # type: ignore[no-redef]
-        """Minimal stub matching openenv.core.env_client.EnvClient."""
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        self.close()
 
-        def __init__(self, *args: Any, **kwargs: Any) -> None:
-            pass
+    def close(self) -> None:
+        close = getattr(self._client, "close", None)
+        if callable(close):
+            close()
 
-from open_range.server.models import RangeAction, RangeObservation, RangeState
+    def reset(self, **kwargs: Any) -> StepResult[RangeObservation]:
+        return self._client.reset(**kwargs)
+
+    def step(self, action: RangeAction, **kwargs: Any) -> StepResult[RangeObservation]:
+        return self._client.step(action, **kwargs)
+
+    def state(self) -> RangeState:
+        return self._client.state()
 
 
 class OpenRangeEnv(EnvClient[RangeAction, RangeObservation, RangeState]):
     """Typed OpenEnv client that speaks the standard reset/step/state contract."""
 
-    def sync(self) -> "OpenRangeEnv":
-        """Compatibility wrapper matching the documented OpenEnv sync pattern."""
-        return self
+    def sync(self) -> Any:
+        """Return the native sync wrapper when available, else a thin proxy."""
+        base_sync = getattr(super(), "sync", None)
+        if callable(base_sync):
+            return base_sync()
+        return _SyncOpenRangeEnv(self)
 
     def _step_payload(self, action: RangeAction) -> dict:
         return {"command": action.command, "mode": action.mode}

src/open_range/models.py

@@ -0,0 +1,44 @@
+"""Shared OpenEnv data models for OpenRange.
+
+These models are intentionally defined outside ``server/`` so both the client
+and server depend on the same shared contract without crossing the client/server
+boundary encouraged by OpenEnv.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import Field
+
+from openenv.core.env_server.types import Action, Observation, State
+
+
+class RangeAction(Action):
+    """Command action for either the Red or Blue operator."""
+
+    command: str
+    mode: Literal["red", "blue"]
+
+
+class RangeObservation(Observation):
+    """Command/result observation for a range step."""
+
+    stdout: str = ""
+    stderr: str = ""
+    flags_captured: list[str] = Field(default_factory=list)
+    alerts: list[str] = Field(default_factory=list)
+
+
+class RangeState(State):
+    """Mutable episode state exposed through the OpenEnv state endpoint."""
+
+    mode: str = ""
+    flags_found: list[str] = Field(default_factory=list)
+    services_status: dict[str, Any] = Field(default_factory=dict)
+    tier: int = 1
+    active_sessions: dict[str, str] = Field(default_factory=dict)
+    auth_attempts: list[dict[str, Any]] = Field(default_factory=list)
+    access_grants: list[str] = Field(default_factory=list)
+    pivot_history: list[dict[str, str]] = Field(default_factory=list)
+    milestones_completed: list[str] = Field(default_factory=list)

src/open_range/protocols.py

@@ -200,7 +200,7 @@ class NPCPersona(BaseModel):
     security_awareness: float = 0.5  # 0.0-1.0
     susceptibility: dict[str, float] = Field(default_factory=dict)
     routine: dict[str, Any] = Field(default_factory=dict)
-    accounts: dict[str, str] = Field(default_factory=dict)
+    accounts: dict[str, Any] = Field(default_factory=dict)
 
 
 class NPCTrafficSpec(BaseModel):

src/open_range/resolve.py

@@ -24,9 +24,9 @@ DEFAULT_CHECKS: list[dict[str, Any]] = [
     {"class": "open_range.validator.build_boot.BuildBootCheck"},
     {"class": "open_range.validator.exploitability.ExploitabilityCheck"},
     {"class": "open_range.validator.patchability.PatchabilityCheck"},
-    {"class": "open_range.validator.evidence.EvidenceSufficiencyCheck"},
+    {"class": "open_range.validator.evidence.EvidenceCheck"},
     {"class": "open_range.validator.reward_grounding.RewardGroundingCheck"},
-    {"class": "open_range.validator.isolation.IsolationLeakageCheck"},
+    {"class": "open_range.validator.isolation.IsolationCheck"},
 ]
 
 

src/open_range/server/app.py

@@ -1,14 +1,9 @@
-"""FastAPI application for OpenRange.
-
-Uses the OpenEnv app factory when openenv is installed, otherwise
-creates a standalone FastAPI app with equivalent endpoints.
-"""
+"""FastAPI application for OpenRange."""
 
 from __future__ import annotations
 
 import logging
-import sys
-import traceback
+import os
 
 from fastapi import FastAPI
 
@@ -16,44 +11,32 @@ logger = logging.getLogger(__name__)
 
 
 def create_app() -> FastAPI:
-    """Create the OpenRange app.
+    """Create the OpenRange app through the canonical OpenEnv factory."""
+    from openenv.core.env_server import create_app as create_openenv_app
 
-    Tries the OpenEnv factory first; falls back to a standalone
-    FastAPI app if openenv is not installed or if the runtime
-    fails to initialise (e.g. missing manifest on HF Spaces).
-    """
+    from open_range.models import RangeAction, RangeObservation
     from open_range.server.environment import RangeEnvironment
-    from open_range.server.models import RangeAction, RangeObservation
 
-    # Try to create the managed runtime (snapshot pool, validator, etc.)
     runtime = None
-    try:
+    runtime_enabled = os.getenv("OPENRANGE_ENABLE_MANAGED_RUNTIME", "").lower() in {
+        "1",
+        "true",
+        "yes",
+    } or bool(os.getenv("OPENRANGE_RUNTIME_MANIFEST"))
+    if runtime_enabled:
         from open_range.server.runtime import ManagedSnapshotRuntime
+
         runtime = ManagedSnapshotRuntime.from_env()
-    except Exception:
-        logger.warning(
-            "ManagedSnapshotRuntime.from_env() failed — running without managed snapshots:\n%s",
-            traceback.format_exc(),
-        )
 
     def env_factory() -> RangeEnvironment:
         return RangeEnvironment(runtime=runtime)
 
-    # Try OpenEnv factory first
-    try:
-        from openenv.core.env_server import create_app as create_openenv_app
-        fastapp = create_openenv_app(
-            env_factory,
-            RangeAction,
-            RangeObservation,
-            env_name="open_range",
-        )
-    except Exception:
-        logger.warning(
-            "OpenEnv create_app failed — creating standalone FastAPI:\n%s",
-            traceback.format_exc(),
-        )
-        fastapp = _create_standalone_app(env_factory)
+    fastapp = create_openenv_app(
+        env_factory,
+        RangeAction,
+        RangeObservation,
+        env_name="open_range",
+    )
 
     fastapp.state.env = env_factory()
     if runtime is not None:
@@ -70,75 +53,13 @@ def create_app() -> FastAPI:
     return fastapp
 
 
-def _create_standalone_app(
-    env_factory: object,
-) -> FastAPI:
-    """Standalone FastAPI app with OpenEnv-compatible endpoints.
-
-    Used when the openenv package is not available.
-    """
-    from open_range.server.models import RangeAction, RangeObservation
-
-    fastapp = FastAPI(title="OpenRange", version="0.1.0")
-    _env_holder: dict = {}
-
-    def _get_env():
-        if "env" not in _env_holder:
-            _env_holder["env"] = env_factory()  # type: ignore[operator]
-        return _env_holder["env"]
-
-    @fastapp.get("/health")
-    def health():
-        return {"status": "healthy"}
-
-    @fastapp.get("/metadata")
-    def metadata():
-        env = _get_env()
-        return env.get_metadata()
-
-    @fastapp.post("/reset")
-    def reset(seed: int | None = None, episode_id: str | None = None):
-        env = _get_env()
-        obs = env.reset(seed=seed, episode_id=episode_id)
-        return {"observation": obs.model_dump()}
-
-    @fastapp.post("/step")
-    def step(action: RangeAction):
-        env = _get_env()
-        obs = env.step(action)
-        return {
-            "observation": obs.model_dump(),
-            "reward": obs.reward,
-            "done": obs.done,
-        }
-
-    @fastapp.get("/state")
-    def state():
-        env = _get_env()
-        return env.state.model_dump()
-
-    return fastapp
-
-
 def main() -> None:
     """Run the installed package entrypoint via uvicorn."""
     import uvicorn
     uvicorn.run("open_range.server.app:app", host="0.0.0.0", port=8000)
 
 
-# Module-level app creation with error reporting
-try:
-    app = create_app()
-except Exception:
-    # If create_app fails entirely, print the error and create a minimal
-    # health-only app so HF Spaces doesn't show "no logs".
-    traceback.print_exc()
-    print("[app.py] FATAL: create_app() failed. Creating minimal health endpoint.", file=sys.stderr)
-    app = FastAPI(title="OpenRange (degraded)")
-
-    @app.get("/health")
-    def _health():
-        return {"status": "degraded", "error": "App failed to initialize"}
+app = create_app()
 
 
 if __name__ == "__main__":

src/open_range/server/environment.py

@@ -43,7 +43,6 @@ DEFAULT_MAX_STEPS = 100
 # Timeout for individual docker exec calls (seconds)
 EXEC_TIMEOUT = 30.0
 
-
 def _extract_command_name(command: str) -> str:
     """Extract the base command name from a full command string."""
     stripped = command.strip()
@@ -59,7 +58,6 @@ def _extract_command_name(command: str) -> str:
             return part.rsplit("/", 1)[-1]
     return parts[0] if parts else ""
 
-
 class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
     """OpenEnv Environment subclass for the cybersecurity range.
 
@@ -476,19 +474,10 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
     # Service lifecycle (subprocess mode)
     # -----------------------------------------------------------------
 
-    # Daemon names to kill when stopping services (legacy + modern).
-    _LEGACY_STOP_DAEMONS = [
-        "nginx", "mysqld", "mariadbd", "slapd", "rsyslogd",
-        "smbd", "postfix", "sshd", "redis-server", "postgres",
-        "jenkins", "prometheus", "grafana-server", "openvpn",
-    ]
-
     def _stop_services(self) -> None:
         """Stop services started by a previous episode.
 
-        Uses the snapshot's ``services`` list when available to determine
-        which daemon names to kill.  Falls back to a legacy kill-list
-        when no snapshot is loaded or the snapshot has no ``services``.
+        Derives daemon names from the snapshot's ``services`` list.
         """
         if self._execution_mode != "subprocess":
             return
@@ -504,27 +493,24 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             except Exception as exc:
                 logger.debug("Failed to stop PID %d: %s", pid, exc)
 
-        # Determine daemon names to kill (from snapshot or legacy list)
         daemon_names: list[str] = []
         if self._snapshot and self._snapshot.services:
             for svc in self._snapshot.services:
                 name = svc.daemon.split("/")[-1].split()[0]
-                if name:
+                if name and name not in daemon_names:
                     daemon_names.append(name)
-        if not daemon_names:
-            daemon_names = list(self._LEGACY_STOP_DAEMONS)
 
-        # Also stop known service processes by name (catches orphans)
-        kill_expr = " ".join(
-            f"pkill -x {name} 2>/dev/null || true;" for name in daemon_names
-        )
-        try:
-            sp.run(
-                ["bash", "-c", kill_expr],
-                capture_output=True, timeout=5,
-            )
-        except Exception:
-            pass
+        for daemon_name in daemon_names:
+            try:
+                sp.run(
+                    ["pkill", "-x", daemon_name],
+                    capture_output=True,
+                    timeout=5,
+                    text=True,
+                    check=False,
+                )
+            except Exception as exc:
+                logger.debug("Failed to stop daemon %s: %s", daemon_name, exc)
 
         self._service_pids = []
         logger.info("Stopped previous episode services")
@@ -532,11 +518,8 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
     def _start_snapshot_services(self, snapshot: SnapshotSpec) -> None:
         """Start services based on snapshot spec (subprocess mode only).
 
-        If the snapshot has a ``services`` list (populated by the Renderer
-        via :func:`generate_service_specs`), each :class:`ServiceSpec` is
-        started generically.  Otherwise falls back to
-        :meth:`_start_services_legacy` which generates ephemeral specs
-        from the topology host names.
+        The snapshot's ``services`` list is normally populated by the Renderer.
+        Older snapshots fall back to topology-derived service specs.
         """
         if self._execution_mode != "subprocess":
             return
@@ -544,7 +527,7 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         if snapshot.services:
             self._start_services_from_specs(snapshot.services)
         else:
-            self._start_services_legacy(snapshot)
+            logger.info("No service specs in snapshot -- skipping service provisioning")
 
     def _start_services_from_specs(self, services: list[ServiceSpec]) -> None:
         """Start a list of :class:`ServiceSpec` entries generically."""
@@ -581,16 +564,42 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         env = os.environ.copy()
         env.update(svc.env_vars)
 
-        # Create log directory
-        if svc.log_dir:
-            os.makedirs(svc.log_dir, exist_ok=True)
+        original_log_dir = svc.log_dir or "/var/log/siem"
+        log_dir = original_log_dir
+        try:
+            os.makedirs(log_dir, exist_ok=True)
+        except PermissionError:
+            if original_log_dir.startswith("/var/log/"):
+                log_dir = os.path.join(
+                    "/tmp/openrange",
+                    original_log_dir.removeprefix("/var/log/"),
+                )
+            else:
+                log_dir = os.path.join("/tmp/openrange", original_log_dir.strip("/"))
+            os.makedirs(log_dir, exist_ok=True)
+
+        init_commands = [
+            cmd.replace(original_log_dir, log_dir)
+            if original_log_dir and original_log_dir != log_dir
+            else cmd
+            for cmd in svc.init_commands
+        ]
+        start_command = (
+            svc.start_command.replace(original_log_dir, log_dir)
+            if original_log_dir and original_log_dir != log_dir
+            else svc.start_command
+        )
 
         # Run init commands
-        for cmd in svc.init_commands:
+        for cmd in init_commands:
             try:
                 result = sp.run(
                     ["bash", "-c", cmd],
-                    capture_output=True, timeout=30, text=True, env=env,
+                    capture_output=True,
+                    timeout=30,
+                    text=True,
+                    env=env,
+                    check=False,
                 )
                 if result.returncode != 0 and result.stderr:
                     logger.debug(
@@ -603,8 +612,12 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         # Start the daemon
         try:
             result = sp.run(
-                ["bash", "-c", svc.start_command],
-                capture_output=True, timeout=30, text=True, env=env,
+                ["bash", "-c", start_command],
+                capture_output=True,
+                timeout=30,
+                text=True,
+                env=env,
+                check=False,
             )
             if result.returncode != 0 and result.stderr:
                 logger.debug(
@@ -625,7 +638,7 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             logger.info("  %s: started (no readiness check)", svc.daemon)
             return
 
-        max_attempts = int(check.timeout_s / max(check.interval_s, 0.1))
+        max_attempts = max(int(check.timeout_s / max(check.interval_s, 0.1)), 1)
         for attempt in range(max_attempts):
             if self._probe_readiness(check):
                 logger.info("  %s: ready (%ds)", svc.daemon, attempt + 1)
@@ -660,53 +673,28 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             pass
         return False
 
-    def _start_services_legacy(self, snapshot: SnapshotSpec) -> None:
-        """Fallback: generate ephemeral ServiceSpecs from topology host names.
-
-        Used when ``snapshot.services`` is empty (old snapshots or manually
-        constructed specs).  Delegates to :func:`generate_service_specs`
-        from the service manifest module.
-        """
-        from open_range.builder.service_manifest import generate_service_specs
-
-        topology = snapshot.topology if isinstance(snapshot.topology, dict) else {}
-        hosts = topology.get("hosts", [])
-        if not hosts:
-            logger.info("No hosts in topology — skipping service provisioning")
-            return
-
-        compose = snapshot.compose if isinstance(snapshot.compose, dict) else {}
-        specs = generate_service_specs(compose=compose, topology=topology)
-
-        if specs:
-            logger.info(
-                "Generated %d ephemeral service specs from topology (legacy path)",
-                len(specs),
-            )
-            self._start_services_from_specs(specs)
-        else:
-            logger.info("No service specs generated from topology")
-
     def _capture_service_pids(self) -> None:
         """Capture PIDs of running service processes."""
-        try:
-            result = sp.run(
-                ["bash", "-c",
-                 "pgrep -x 'nginx|mysqld|mariadbd|slapd|rsyslogd|smbd|sshd"
-                 "|redis-server|postgres|jenkins|prometheus|grafana-server"
-                 "|openvpn' 2>/dev/null || true"],
-                capture_output=True, timeout=5, text=True,
-            )
-            for line in result.stdout.strip().split("\n"):
-                line = line.strip()
-                if line.isdigit():
-                    self._service_pids.append(int(line))
-        except Exception:
-            pass
+        self._service_pids = []
+        daemon_names: list[str] = []
+        if self._snapshot and self._snapshot.services:
+            for svc in self._snapshot.services:
+                name = svc.daemon.split("/")[-1].split()[0]
+                if name and name not in daemon_names:
+                    daemon_names.append(name)
 
-    # -----------------------------------------------------------------
-    # NPC lifecycle
-    # -----------------------------------------------------------------
+        for daemon_name in daemon_names:
+            try:
+                result = sp.run(
+                    ["pgrep", "-x", daemon_name],
+                    capture_output=True, timeout=5, text=True, check=False,
+                )
+            except Exception:
+                continue
+            for line in result.stdout.splitlines():
+                pid = line.strip()
+                if pid.isdigit():
+                    self._service_pids.append(int(pid))
 
     def _build_container_set(self) -> "ContainerSet | None":
         """Build a ContainerSet from running Docker containers.
@@ -869,8 +857,6 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             self._snapshot_id = admitted.snapshot_id
             snap = admitted.snapshot
         else:
-            # Backward-compatible minimal stub for tests, demos, and local
-            # mock-mode usage when a managed runtime is not configured.
             self._snapshot_id = None
             snap = SnapshotSpec(
                 topology={"hosts": ["attacker", "siem"]},
@@ -1120,47 +1106,45 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         """Determine which container to route the command to.
 
         Reads from the snapshot topology to find the appropriate host:
-        - Red: host with ``role: "attacker"`` or ``zone: "external"``.
-        - Blue: host with ``role: "siem"`` or ``zone: "management"``.
+        - Red: host with role=attacker or zone=external.
+        - Blue: host with role=siem or zone=management.
 
-        Falls back to ``"attacker"``/``"siem"`` if no snapshot is loaded
-        or no matching host is found in the topology.
+        The snapshot topology must define hosts with roles or zones.
+        For string-only host lists, matches by name then falls back to
+        positional convention (first host for Red, last for Blue).
         """
-        red_default = "attacker"
-        blue_default = "siem"
-
-        if self._snapshot and isinstance(self._snapshot.topology, dict):
-            hosts = self._snapshot.topology.get("hosts", [])
+        if not self._snapshot or not isinstance(self._snapshot.topology, dict):
+            raise RuntimeError("Cannot resolve target — no snapshot topology loaded")
 
-            if action.mode == "red":
-                # Look for a host with role "attacker" or zone "external"
-                for h in hosts:
-                    if isinstance(h, dict):
-                        if h.get("role") == "attacker" or h.get("zone") == "external":
-                            host_name = h.get("name", h.get("hostname", red_default))
-                            return self._container_name(host_name)
-                # Fallback: check if "attacker" is in the hosts list (string entries)
-                for h in hosts:
-                    if isinstance(h, str) and h == "attacker":
-                        return self._container_name("attacker")
-                # Last resort
-                return self._container_name(red_default)
-            else:
-                # Look for a host with role "siem" or zone "management"
-                for h in hosts:
-                    if isinstance(h, dict):
-                        if h.get("role") == "siem" or h.get("zone") == "management":
-                            host_name = h.get("name", h.get("hostname", blue_default))
-                            return self._container_name(host_name)
-                # Fallback: check if "siem" is in the hosts list (string entries)
-                for h in hosts:
-                    if isinstance(h, str) and h == "siem":
-                        return self._container_name("siem")
-                # Last resort
-                return self._container_name(blue_default)
-
-        # No snapshot loaded — use hardcoded defaults as last resort
-        return self._container_name(red_default if action.mode == "red" else blue_default)
+        hosts = self._snapshot.topology.get("hosts", [])
+        if not hosts:
+            raise RuntimeError("Cannot resolve target — snapshot topology has no hosts")
+
+        target_role = "attacker" if action.mode == "red" else "siem"
+        target_zone = "external" if action.mode == "red" else "management"
+
+        # Look for a host with matching role or zone
+        for h in hosts:
+            if isinstance(h, dict):
+                if h.get("role") == target_role or h.get("zone") == target_zone:
+                    host_name = h.get("name", h.get("hostname", ""))
+                    if host_name:
+                        return self._container_name(host_name)
+
+        # String host list: match by name
+        for h in hosts:
+            name = h if isinstance(h, str) else h.get("name", "")
+            if name == target_role:
+                return self._container_name(name)
+
+        # Use positional convention: first host for Red, last for Blue
+        fallback = hosts[0] if action.mode == "red" else hosts[-1]
+        name = fallback if isinstance(fallback, str) else fallback.get("name", fallback.get("hostname", ""))
+        logger.warning(
+            "No host with role=%s or zone=%s found; using positional fallback: %s",
+            target_role, target_zone, name,
+        )
+        return self._container_name(name)
 
     # -----------------------------------------------------------------
     # Core API
@@ -1285,6 +1269,20 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
         Returns:
             RangeObservation with command output and reward.
         """
+        if self._snapshot is None:
+            self._snapshot = self._select_snapshot(**kwargs)
+            tier = self._snapshot.topology.get("tier", 1) if isinstance(
+                self._snapshot.topology, dict
+            ) else 1
+            self._state = RangeState(
+                episode_id=self._state.episode_id or str(uuid4()),
+                step_count=0,
+                mode=action.mode,
+                flags_found=list(self._state.flags_found),
+                services_status=dict(self._state.services_status),
+                tier=tier,
+            )
+
         self._state.step_count += 1
         self._state.mode = action.mode
 
@@ -1332,7 +1330,6 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             self._report_if_done(obs)
             return obs
 
-
         # Route to container
         target = self._resolve_target(action)
         timeout = timeout_s or self._exec_timeout
@@ -1516,16 +1513,7 @@ class RangeEnvironment(Environment[RangeAction, RangeObservation, RangeState]):
             if siem_alerts:
                 return siem_alerts
 
-        # Synthetic fallback: treat ALL Red actions as potential alerts
-        alerts: list[str] = []
-        for record in self._red_history:
-            cmd = record.get("cmd_name", "")
-            if cmd:
-                alerts.append(
-                    f"[IDS] Suspicious activity detected: {cmd} "
-                    f"at step {record['step']}"
-                )
-        return alerts
+        return []
 
     # -----------------------------------------------------------------
     # Introspection (for reward computation and debugging)

src/open_range/server/models.py

@@ -1,60 +1,5 @@
-"""OpenEnv-compatible models for OpenRange.
+"""Compatibility re-export for code still importing from ``open_range.server``."""
 
-RangeAction, RangeObservation, and RangeState extend the OpenEnv base
-types. Falls back to Pydantic stubs if openenv is not installed.
-"""
+from open_range.models import RangeAction, RangeObservation, RangeState
 
-from __future__ import annotations
-
-from typing import Any, Literal
-
-from pydantic import Field
-
-try:
-    from openenv.core.env_server.types import Action, Observation, State
-except ImportError:
-    from pydantic import BaseModel, ConfigDict
-
-    class Action(BaseModel):  # type: ignore[no-redef]
-        model_config = ConfigDict(extra="forbid", validate_assignment=True)
-        metadata: dict[str, Any] = Field(default_factory=dict)
-
-    class Observation(BaseModel):  # type: ignore[no-redef]
-        model_config = ConfigDict(extra="forbid", validate_assignment=True)
-        done: bool = False
-        reward: bool | int | float | None = None
-        metadata: dict[str, Any] = Field(default_factory=dict)
-
-    class State(BaseModel):  # type: ignore[no-redef]
-        model_config = ConfigDict(extra="allow")
-        episode_id: str | None = None
-        step_count: int = Field(default=0, ge=0)
-
-
-class RangeAction(Action):
-    command: str
-    mode: Literal["red", "blue"]
-
-
-class RangeObservation(Observation):
-    # done and reward inherited from Observation
-    stdout: str = ""
-    stderr: str = ""
-    flags_captured: list[str] = Field(default_factory=list)
-    alerts: list[str] = Field(default_factory=list)
-
-
-class RangeState(State):
-    # episode_id and step_count inherited from State
-    mode: str = ""
-    flags_found: list[str] = Field(default_factory=list)
-    services_status: dict[str, Any] = Field(default_factory=dict)
-    tier: int = 1
-    # Auth scenario (#25): session tracking
-    active_sessions: dict[str, str] = Field(default_factory=dict)  # host -> username
-    auth_attempts: list[dict[str, Any]] = Field(default_factory=list)
-    # Pivot mechanics (#26): access and lateral movement tracking
-    access_grants: list[str] = Field(default_factory=list)  # ["host:service", ...]
-    pivot_history: list[dict[str, str]] = Field(default_factory=list)  # [{from: "web", to: "db", via: "credential_reuse"}]
-    # Task engine (#17): milestone tracking
-    milestones_completed: list[str] = Field(default_factory=list)
+__all__ = ["RangeAction", "RangeObservation", "RangeState"]

src/open_range/server/rewards.py

@@ -19,7 +19,7 @@ from typing import Any
 
 from open_range.protocols import SnapshotSpec
 
-from open_range.server.models import RangeAction, RangeObservation, RangeState
+from open_range.models import RangeAction, RangeObservation, RangeState
 
 logger = logging.getLogger(__name__)
 
@@ -186,8 +186,13 @@ class CompositeRedReward:
         npc_compromised: bool = False,
         hallucinated_count: int = 0,
         tier: int = 1,
+        valid_flags: set[str] | None = None,
+        snapshot: SnapshotSpec | None = None,
     ) -> float:
-        valid_flags: set[str] = set()  # caller should supply if known
+        if valid_flags is None and snapshot is not None:
+            valid_flags = {f.value for f in snapshot.flags}
+        if valid_flags is None:
+            valid_flags = set()
         total = 0.0
         if submitted_flag is not None:
             total += self.weights["flag"] * self.flag.score(submitted_flag, valid_flags)
@@ -237,10 +242,12 @@ class CompositeRedReward:
         # Evidence
         evidence_entries = [r for r in red_history if r.get("type") == "evidence"]
         evidence_content = evidence_entries[-1].get("content", "") if evidence_entries else ""
-        topo_hosts = {
-            h.get("name", "") if isinstance(h, dict) else ""
-            for h in snapshot.topology.get("hosts", [])
-        }
+        topo_hosts: set[str] = set()
+        if isinstance(snapshot.topology, dict):
+            topo_hosts = {
+                h.get("name", "") if isinstance(h, dict) else ""
+                for h in snapshot.topology.get("hosts", [])
+            }
         evidence_score = self.evidence.score(evidence_content, topo_hosts)
 
         # Social engineering -- reactive NPC actions from send_phish or

src/open_range/server/runtime.py

@@ -38,7 +38,7 @@ from open_range.protocols import (
     SnapshotSpec,
 )
 from open_range.server.compose_runner import BootedSnapshotProject, ComposeProjectRunner
-from open_range.server.models import RangeState
+from open_range.models import RangeState
 from open_range.validator.build_boot import BuildBootCheck
 from open_range.validator.difficulty import DifficultyCheck
 from open_range.validator.evidence import EvidenceCheck

src/open_range/training/runner.py

@@ -200,7 +200,7 @@ class CurriculumRunner:
         self, manifest_path: str, seed: int, episode_num: int
     ) -> EpisodeRecord:
         """Run a single episode and return an EpisodeRecord."""
-        from open_range.server.models import RangeAction
+        from open_range.models import RangeAction
 
         start = time.time()
 

src/open_range/training/synthetic.py

@@ -22,7 +22,7 @@ from open_range.agents.replay_agent import ScriptedBlueAgent, ScriptedRedAgent
 from open_range.builder.builder import LLMSnapshotBuilder, TemplateOnlyBuilder
 from open_range.protocols import BuildContext, SnapshotBuilder, SnapshotSpec, Vulnerability
 from open_range.server.environment import RangeEnvironment
-from open_range.server.models import RangeAction, RangeObservation
+from open_range.models import RangeAction, RangeObservation
 from open_range.training.trajectory import TrajectoryLogger
 
 logger = logging.getLogger(__name__)

src/open_range/validator/evidence.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import shlex
+
 from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
 
 
@@ -30,16 +32,17 @@ class EvidenceCheck:
                 host, path = "siem", loc
 
             try:
+                safe_path = shlex.quote(path)
                 if item.type in ("log_entry", "alert"):
                     # grep for pattern in the file
-                    cmd = f"grep -c '{pattern}' {path}" if pattern else f"test -f {path} && echo ok"
+                    cmd = f"grep -c {shlex.quote(pattern)} {safe_path}" if pattern else f"test -f {safe_path} && echo ok"
                     output = await containers.exec(host, cmd)
                     # grep -c returns "0" if no matches — that means missing
                     if pattern and output.strip() in ("0", ""):
                         missing.append({"item": item.type, "location": loc, "pattern": pattern})
                 else:
                     # file existence check
-                    output = await containers.exec(host, f"test -f {path} && echo exists")
+                    output = await containers.exec(host, f"test -f {safe_path} && echo exists")
                     if "exists" not in output:
                         missing.append({"item": item.type, "location": loc})
             except Exception as exc:  # noqa: BLE001

src/open_range/validator/exploitability.py

@@ -2,9 +2,13 @@
 
 from __future__ import annotations
 
+import logging
+
 from open_range.protocols import CheckResult, ContainerSet, SnapshotSpec
 from open_range.validator._golden_path import execute_step
 
+logger = logging.getLogger(__name__)
+
 _META_COMMANDS = {"submit_flag", "submit_evidence", "submit_finding", "auth", "logout"}
 
 
@@ -21,6 +25,7 @@ class ExploitabilityCheck:
 
         failed_steps: list[dict] = []
         skipped_steps: list[int] = []
+        unvalidated_steps: list[int] = []
         for step in snapshot.golden_path:
             cmd_name = step.command.strip().split()[0] if step.command.strip() else ""
             if cmd_name in _META_COMMANDS:
@@ -37,7 +42,14 @@ class ExploitabilityCheck:
                 continue
 
             expected = step.expect_in_stdout
-            if expected and expected not in output:
+            if not expected:
+                logger.warning(
+                    "exploitability: golden path step %d has no expect_in_stdout — "
+                    "output not validated",
+                    step.step,
+                )
+                unvalidated_steps.append(step.step)
+            elif expected not in output:
                 failed_steps.append({
                     "step": step.step,
                     "expected": expected,
@@ -45,12 +57,19 @@ class ExploitabilityCheck:
                 })
 
         passed = len(failed_steps) == 0
+        issues: list[str] = []
+        if unvalidated_steps:
+            issues.append(
+                f"Steps with no expected output validation: {unvalidated_steps}"
+            )
         return CheckResult(
             name="exploitability",
             passed=passed,
             details={
                 "failed_steps": failed_steps,
                 "skipped_steps": skipped_steps,
+                "unvalidated_steps": unvalidated_steps,
+                "issues": issues,
                 "total_steps": len(snapshot.golden_path),
             },
             error="" if passed else f"{len(failed_steps)} golden-path step(s) failed",

src/open_range/validator/patchability.py

@@ -97,16 +97,20 @@ class PatchabilityCheck:
         tested_count = 0
 
         for vuln in vulns:
-            # --- Skip if no remediation defined ---
-            if not vuln.remediation:
-                results.append({"vuln": vuln.id, "skipped": "no remediation defined"})
+            # --- Fail if no remediation defined ---
+            if not vuln.remediation or not vuln.remediation.strip():
+                msg = "no remediation defined"
+                logger.warning("patchability: vuln %s has %s — counting as failure", vuln.id, msg)
+                results.append({"vuln": vuln.id, "passed": False, "reason": msg})
+                all_ok = False
                 continue
 
-            # --- Skip non-executable remediation (prose) ---
+            # --- Fail non-executable remediation (prose) ---
             if not _looks_executable(vuln.remediation):
                 msg = f"remediation is not executable: {vuln.remediation!r}"
-                logger.warning("patchability: skipping vuln %s — %s", vuln.id, msg)
-                results.append({"vuln": vuln.id, "skipped": msg})
+                logger.warning("patchability: vuln %s — %s — counting as failure", vuln.id, msg)
+                results.append({"vuln": vuln.id, "passed": False, "reason": msg})
+                all_ok = False
                 continue
 
             # Find the golden-path step(s) that exercise this vuln.

src/open_range/validator/task_feasibility.py

@@ -25,10 +25,19 @@ class TaskFeasibilityCheck:
                 topo_hosts.add(str(h))
         topo_hosts.discard("")
 
+        # Fail early if topology has no hosts.
+        if not topo_hosts:
+            return CheckResult(
+                name="task_feasibility",
+                passed=False,
+                details={"issues": ["Topology has no hosts defined"]},
+                error="Topology has no hosts defined",
+            )
+
         # 1. Golden-path hosts exist in topology.
         for step in snapshot.golden_path:
             host = getattr(step, "host", None) or "attacker"
-            if host not in topo_hosts and topo_hosts:
+            if host not in topo_hosts:
                 issues.append(f"golden path step {step.step}: host '{host}' not in topology")
 
         # 2. Evidence targets reference existing containers.
@@ -38,7 +47,7 @@ class TaskFeasibilityCheck:
                 host = loc.split(":")[0]
             else:
                 host = "siem"
-            if host not in topo_hosts and topo_hosts:
+            if host not in topo_hosts:
                 issues.append(f"evidence item '{item.type}' references unknown host '{host}'")
 
         # 3. Exploit chain vuln IDs exist in truth_graph.
@@ -49,7 +58,7 @@ class TaskFeasibilityCheck:
 
         # 4. Flag hosts exist in topology.
         for flag in snapshot.flags:
-            if flag.host not in topo_hosts and topo_hosts:
+            if flag.host not in topo_hosts:
                 issues.append(f"flag '{flag.id}' references unknown host '{flag.host}'")
 
         passed = len(issues) == 0

start.sh

@@ -6,10 +6,10 @@
 # NOT called at container boot — the Dockerfile starts only uvicorn.
 #
 # Usage:  start.sh <snapshot_dir>
-#   snapshot_dir must contain a spec.json with a topology.hosts list.
-#   Each host name maps to a known service (nginx, mysql, slapd, etc.).
+#   snapshot_dir must contain a spec.json.
 #
-# Services are started based on what the snapshot requires, not hardcoded.
+# If spec.json contains a "services" list (ServiceSpec entries), those are
+# started generically.  Otherwise falls back to legacy host-name mapping.
 # =============================================================================
 
 set -uo pipefail
@@ -34,7 +34,7 @@ cleanup() {
 # service lifecycle via _stop_services() / _start_snapshot_services().
 trap cleanup INT TERM
 
-# ── Parse snapshot topology ───────────────────────────────────────────────────
+# ── Parse snapshot ────────────────────────────────────────────────────────────
 
 mkdir -p "${CONSOLIDATED}"
 
@@ -43,6 +43,107 @@ if [ ! -f "${SNAPSHOT_DIR}/spec.json" ]; then
     exit 1
 fi
 
+# ── Check for declarative services list ───────────────────────────────────────
+# If spec.json contains "services" entries (ServiceSpec), start them generically
+# via Python. This is the modern path populated by the Renderer.
+
+HAS_SERVICES=$(python3 -c "
+import json
+with open('${SNAPSHOT_DIR}/spec.json') as f:
+    spec = json.load(f)
+svcs = spec.get('services', [])
+print(len(svcs))
+" 2>/dev/null || echo "0")
+
+if [ "$HAS_SERVICES" -gt 0 ] 2>/dev/null; then
+    echo "[start.sh] Found $HAS_SERVICES declared service(s) — using spec-driven startup"
+
+    python3 -c "
+import json, subprocess, sys, time, os, socket
+
+with open('${SNAPSHOT_DIR}/spec.json') as f:
+    spec = json.load(f)
+
+pids = []
+for svc in spec.get('services', []):
+    daemon = svc.get('daemon', '')
+    host = svc.get('host', '')
+    print(f'[start.sh] Starting service: {daemon} (host={host})')
+
+    env = os.environ.copy()
+    env.update(svc.get('env_vars', {}))
+
+    log_dir = svc.get('log_dir', '')
+    if log_dir:
+        os.makedirs(log_dir, exist_ok=True)
+
+    # Init commands
+    for cmd in svc.get('init_commands', []):
+        try:
+            subprocess.run(['bash', '-c', cmd], capture_output=True, timeout=30, env=env)
+        except Exception as e:
+            print(f'[start.sh]   init warning: {e}', file=sys.stderr)
+
+    # Start command
+    start_cmd = svc.get('start_command', '')
+    if start_cmd:
+        try:
+            subprocess.run(['bash', '-c', start_cmd], capture_output=True, timeout=30, env=env)
+        except Exception as e:
+            print(f'[start.sh]   start warning: {e}', file=sys.stderr)
+
+    # Readiness
+    readiness = svc.get('readiness', {})
+    rtype = readiness.get('type', 'tcp')
+    timeout_s = readiness.get('timeout_s', 30)
+    interval_s = readiness.get('interval_s', 1.0)
+    port = readiness.get('port', 0)
+    url = readiness.get('url', '')
+    command = readiness.get('command', '')
+
+    if (rtype == 'tcp' and port == 0 and not url and not command):
+        print(f'[start.sh]   {daemon}: started (no readiness check)')
+        continue
+
+    max_attempts = int(timeout_s / max(interval_s, 0.1))
+    ready = False
+    for attempt in range(max_attempts):
+        try:
+            if rtype == 'tcp' and port > 0:
+                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                s.settimeout(2)
+                s.connect(('127.0.0.1', port))
+                s.close()
+                ready = True
+            elif rtype == 'http' and url:
+                r = subprocess.run(['curl', '-sf', url], capture_output=True, timeout=3)
+                ready = (r.returncode == 0)
+            elif rtype == 'command' and command:
+                r = subprocess.run(['bash', '-c', command], capture_output=True, timeout=5)
+                ready = (r.returncode == 0)
+        except Exception:
+            pass
+        if ready:
+            print(f'[start.sh]   {daemon}: ready ({attempt + 1}s)')
+            break
+        time.sleep(interval_s)
+    else:
+        if not ready:
+            print(f'[start.sh]   {daemon}: readiness timeout after {timeout_s}s')
+"
+
+    echo "============================================================"
+    echo "[start.sh] Spec-driven services started."
+    echo "[start.sh] Logs at: ${LOGDIR}/"
+    echo "============================================================"
+    exit 0
+fi
+
+# ── Legacy fallback: host-name-based service mapping ──────────────────────────
+# Used when spec.json has no "services" list (old snapshots).
+
+echo "[start.sh] No declared services — falling back to legacy host mapping"
+
 # Extract host list from topology
 HOSTS=$(python3 -c "
 import json, sys

tests/test_builder.py

@@ -188,9 +188,10 @@ async def test_mutator_compiles_root_snapshot_from_manifest_graph(tier1_manifest
     assert topology["dependency_edges"]
     assert topology["trust_edges"]
     assert "principal_catalog" in topology
-    assert "schen" in topology["principal_catalog"]
-    assert "schen" not in {user["username"] for user in topology["users"]}
-    assert topology["manifest_normalization"]["trust_only_principals"]
+    # After fixing tier1_basic.yaml, all trust_relationships reference
+    # users that exist in the users section, so there should be no
+    # trust-only principals.
+    assert not topology["manifest_normalization"]["trust_only_principals"]
 
 
 @pytest.mark.asyncio

tests/test_client.py

@@ -0,0 +1,25 @@
+"""Tests for the typed OpenEnv client."""
+
+from open_range.client.client import OpenRangeEnv
+from open_range.models import RangeAction, RangeObservation, RangeState
+from open_range.server.models import (
+    RangeAction as ServerRangeAction,
+    RangeObservation as ServerRangeObservation,
+    RangeState as ServerRangeState,
+)
+
+
+class TestOpenRangeClient:
+    def test_sync_returns_openenv_sync_wrapper(self):
+        client = OpenRangeEnv(base_url="http://localhost:8000")
+        sync_client = client.sync()
+
+        assert sync_client is not client
+        assert hasattr(sync_client, "reset")
+        assert hasattr(sync_client, "step")
+        assert hasattr(sync_client, "__enter__")
+
+    def test_server_model_module_reexports_shared_models(self):
+        assert ServerRangeAction is RangeAction
+        assert ServerRangeObservation is RangeObservation
+        assert ServerRangeState is RangeState

tests/test_environment.py

@@ -13,6 +13,14 @@ from open_range.protocols import (
 from open_range.server.environment import RangeEnvironment, _extract_command_name
 from open_range.server.models import RangeAction, RangeObservation, RangeState
 
+# Minimal snapshot for tests that just need reset() to work
+_MINIMAL_SNAPSHOT = SnapshotSpec(
+    topology={"hosts": ["attacker", "siem"]},
+    flags=[],
+    golden_path=[],
+    task=TaskSpec(red_briefing="Test mode.", blue_briefing="Test mode."),
+)
+
 
 class TestCommandExtraction:
     """Helper: extracting base command name from full command strings."""
@@ -35,21 +43,21 @@ class TestReset:
 
     def test_reset_returns_observation(self):
         env = RangeEnvironment(docker_available=False)
-        obs = env.reset()
+        obs = env.reset(snapshot=_MINIMAL_SNAPSHOT)
         assert isinstance(obs, RangeObservation)
         assert "Range ready" in obs.stdout
 
     def test_reset_sets_episode_id(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset(episode_id="ep_42")
+        env.reset(snapshot=_MINIMAL_SNAPSHOT, episode_id="ep_42")
         assert env.state.episode_id == "ep_42"
 
     def test_reset_clears_step_count(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         env.step(RangeAction(command="nmap -sV web", mode="red"))
         assert env.state.step_count == 1
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         assert env.state.step_count == 0
 
     def test_reset_with_snapshot(self, sample_snapshot_spec):
@@ -64,14 +72,14 @@ class TestRedStep:
 
     def test_red_step_returns_observation(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         action = RangeAction(command="nmap -sV web", mode="red")
         obs = env.step(action)
         assert isinstance(obs, RangeObservation)
 
     def test_red_step_increments_counter(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         env.step(RangeAction(command="nmap -sV web", mode="red"))
         assert env.state.step_count == 1
         env.step(RangeAction(command="curl http://web", mode="red"))
@@ -80,7 +88,7 @@ class TestRedStep:
     def test_red_any_command_forwarded(self):
         """No artificial allowlist — commands route to the attacker container."""
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         obs = env.step(RangeAction(command="iptables -L", mode="red"))
         # In mock mode, this runs on attacker container (not rejected)
         assert obs.stderr == ""
@@ -88,7 +96,7 @@ class TestRedStep:
 
     def test_red_action_logged(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         env.step(RangeAction(command="nmap -sV web", mode="red"))
         assert len(env.red_history) >= 1
 
@@ -98,20 +106,20 @@ class TestBlueStep:
 
     def test_blue_step_returns_observation(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         obs = env.step(RangeAction(command="tail_log /var/log/syslog", mode="blue"))
         assert isinstance(obs, RangeObservation)
 
     def test_blue_submit_finding(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         obs = env.step(RangeAction(command="submit_finding SQL injection detected", mode="blue"))
         assert "recorded" in obs.stdout.lower() or "submitted" in obs.stdout.lower()
 
     def test_blue_any_command_forwarded(self):
         """No artificial allowlist — commands route to the siem container."""
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         obs = env.step(RangeAction(command="nmap -sV web", mode="blue"))
         # In mock mode, this runs on siem container (not rejected)
         assert obs.stderr == ""
@@ -119,13 +127,13 @@ class TestBlueStep:
 
     def test_blue_empty_command_rejected(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         obs = env.step(RangeAction(command="", mode="blue"))
         assert obs.stderr != ""
 
     def test_step_passes_timeout_override_to_executor(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         seen = {}
 
         def fake_exec(container_name, command, timeout_s=None):
@@ -186,7 +194,7 @@ class TestTermination:
 
     def test_max_steps_terminates(self):
         env = RangeEnvironment(docker_available=False, max_steps=3)
-        env.reset()
+        env.reset(snapshot=_MINIMAL_SNAPSHOT)
         env.step(RangeAction(command="nmap -sV web", mode="red"))
         env.step(RangeAction(command="curl http://web", mode="red"))
         obs = env.step(RangeAction(command="curl http://web/login", mode="red"))
@@ -198,7 +206,7 @@ class TestStateProperty:
 
     def test_state_reflects_episode(self):
         env = RangeEnvironment(docker_available=False)
-        env.reset(episode_id="test_ep")
+        env.reset(snapshot=_MINIMAL_SNAPSHOT, episode_id="test_ep")
         assert env.state.episode_id == "test_ep"
         assert env.state.step_count == 0
         env.step(RangeAction(command="nmap -sV web", mode="red"))

tests/test_service_spec.py

@@ -0,0 +1,597 @@
+"""Tests for ServiceSpec, ReadinessCheck, and generate_service_specs().
+
+Covers:
+- ServiceSpec / ReadinessCheck serialization round-trips
+- generate_service_specs() with compose input (tier-1 and tier-3 services)
+- generate_service_specs() with topology fallback (no compose)
+- Backward compatibility: SnapshotSpec without services field
+- Unknown images produce no specs (graceful skip)
+- Environment service lifecycle integration
+- Renderer generates services field in snapshot
+"""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from open_range.builder.service_manifest import (
+    _HOST_NAME_HINTS,
+    _IMAGE_SERVICE_HINTS,
+    _match_image_hint,
+    generate_service_specs,
+)
+from open_range.protocols import (
+    ReadinessCheck,
+    ServiceSpec,
+    SnapshotSpec,
+    TaskSpec,
+)
+
+
+# ---------------------------------------------------------------------------
+# ServiceSpec / ReadinessCheck serialization
+# ---------------------------------------------------------------------------
+
+
+class TestReadinessCheck:
+    """ReadinessCheck model basics and serialization."""
+
+    def test_defaults(self):
+        rc = ReadinessCheck()
+        assert rc.type == "tcp"
+        assert rc.port == 0
+        assert rc.url == ""
+        assert rc.command == ""
+        assert rc.timeout_s == 30
+        assert rc.interval_s == 1.0
+
+    def test_tcp_check(self):
+        rc = ReadinessCheck(type="tcp", port=80, timeout_s=10)
+        assert rc.type == "tcp"
+        assert rc.port == 80
+
+    def test_http_check(self):
+        rc = ReadinessCheck(type="http", url="http://localhost:8080/health")
+        assert rc.type == "http"
+        assert rc.url == "http://localhost:8080/health"
+
+    def test_command_check(self):
+        rc = ReadinessCheck(type="command", command="pgrep -x nginx")
+        assert rc.type == "command"
+        assert rc.command == "pgrep -x nginx"
+
+    def test_roundtrip_json(self):
+        rc = ReadinessCheck(type="http", url="http://localhost:9090", timeout_s=15)
+        data = rc.model_dump()
+        rc2 = ReadinessCheck(**data)
+        assert rc2.type == rc.type
+        assert rc2.url == rc.url
+        assert rc2.timeout_s == rc.timeout_s
+
+
+class TestServiceSpec:
+    """ServiceSpec model basics and serialization."""
+
+    def test_required_fields(self):
+        svc = ServiceSpec(host="web", daemon="nginx", start_command="nginx &")
+        assert svc.host == "web"
+        assert svc.daemon == "nginx"
+        assert svc.start_command == "nginx &"
+
+    def test_defaults(self):
+        svc = ServiceSpec(host="web", daemon="nginx", start_command="nginx &")
+        assert svc.packages == []
+        assert svc.init_commands == []
+        assert svc.env_vars == {}
+        assert svc.log_dir == ""
+        assert isinstance(svc.readiness, ReadinessCheck)
+
+    def test_full_spec(self):
+        svc = ServiceSpec(
+            host="db",
+            daemon="mysqld",
+            packages=["default-mysql-server"],
+            init_commands=["mkdir -p /var/run/mysqld"],
+            start_command="mysqld --user=mysql &",
+            readiness=ReadinessCheck(
+                type="command",
+                command="mysqladmin ping",
+                timeout_s=30,
+            ),
+            log_dir="/var/log/siem",
+            env_vars={"MYSQL_ROOT_PASSWORD": "secret"},
+        )
+        assert svc.daemon == "mysqld"
+        assert len(svc.init_commands) == 1
+        assert svc.readiness.type == "command"
+        assert svc.env_vars["MYSQL_ROOT_PASSWORD"] == "secret"
+
+    def test_roundtrip_json(self):
+        svc = ServiceSpec(
+            host="web",
+            daemon="nginx",
+            packages=["nginx"],
+            init_commands=["mkdir -p /var/log/nginx"],
+            start_command="nginx -g 'daemon off;' &",
+            readiness=ReadinessCheck(type="tcp", port=80),
+            log_dir="/var/log/siem",
+            env_vars={"SERVER_NAME": "web.corp.local"},
+        )
+        data = json.loads(svc.model_dump_json())
+        svc2 = ServiceSpec(**data)
+        assert svc2.host == svc.host
+        assert svc2.daemon == svc.daemon
+        assert svc2.packages == svc.packages
+        assert svc2.readiness.port == 80
+        assert svc2.env_vars == svc.env_vars
+
+
+# ---------------------------------------------------------------------------
+# SnapshotSpec backward compatibility
+# ---------------------------------------------------------------------------
+
+
+class TestSnapshotSpecServices:
+    """SnapshotSpec.services field: default and serialization."""
+
+    def test_default_empty(self):
+        spec = SnapshotSpec()
+        assert spec.services == []
+
+    def test_with_services(self):
+        spec = SnapshotSpec(
+            topology={"hosts": ["web"]},
+            services=[
+                ServiceSpec(host="web", daemon="nginx", start_command="nginx &"),
+            ],
+        )
+        assert len(spec.services) == 1
+        assert spec.services[0].daemon == "nginx"
+
+    def test_roundtrip_preserves_services(self):
+        svc = ServiceSpec(
+            host="db",
+            daemon="mysqld",
+            start_command="mysqld &",
+            readiness=ReadinessCheck(type="tcp", port=3306),
+        )
+        spec = SnapshotSpec(
+            topology={"hosts": ["db"]},
+            services=[svc],
+        )
+        data = json.loads(spec.model_dump_json())
+        spec2 = SnapshotSpec(**data)
+        assert len(spec2.services) == 1
+        assert spec2.services[0].daemon == "mysqld"
+        assert spec2.services[0].readiness.port == 3306
+
+    def test_old_snapshot_without_services_parses(self):
+        """Simulate loading a JSON snapshot that predates the services field."""
+        old_data = {
+            "topology": {"hosts": ["web", "db"]},
+            "flags": [],
+            "golden_path": [],
+        }
+        spec = SnapshotSpec(**old_data)
+        assert spec.services == []
+
+
+# ---------------------------------------------------------------------------
+# generate_service_specs() — compose input
+# ---------------------------------------------------------------------------
+
+
+class TestGenerateFromCompose:
+    """generate_service_specs() with compose services dict."""
+
+    def test_tier1_basic_compose(self):
+        """Tier 1 compose with common services maps correctly."""
+        compose = {
+            "services": {
+                "web": {"image": "nginx:1.25"},
+                "db": {"image": "mysql:8.0"},
+                "ldap": {"image": "osixia/openldap:1.5"},
+                "siem": {"image": "rsyslog:latest"},
+                "files": {"image": "samba:latest"},
+                "mail": {"image": "postfix:latest"},
+                "attacker": {"image": "kali:latest"},
+            }
+        }
+        topology = {"hosts": ["attacker", "web", "db", "ldap", "siem", "files", "mail"]}
+        specs = generate_service_specs(compose, topology)
+
+        daemon_names = {s.daemon for s in specs}
+        assert "nginx" in daemon_names
+        assert "mysqld" in daemon_names
+        assert "slapd" in daemon_names
+        assert "rsyslogd" in daemon_names
+        assert "smbd" in daemon_names
+        assert "master" in daemon_names  # postfix
+
+    def test_tier3_compose_with_extra_services(self):
+        """Tier 3 compose with redis, postgres, jenkins."""
+        compose = {
+            "services": {
+                "web": {"image": "nginx:1.25"},
+                "cache": {"image": "redis:7"},
+                "db": {"image": "postgres:16"},
+                "ci_cd": {"image": "jenkins/jenkins:lts"},
+                "monitoring": {"image": "prometheus:latest"},
+            }
+        }
+        topology = {"hosts": ["web", "cache", "db", "ci_cd", "monitoring"]}
+        specs = generate_service_specs(compose, topology)
+
+        daemon_names = {s.daemon for s in specs}
+        assert "nginx" in daemon_names
+        assert "redis-server" in daemon_names
+        assert "postgres" in daemon_names
+        assert "java" in daemon_names  # jenkins
+        assert "prometheus" in daemon_names
+
+    def test_unknown_image_skipped(self):
+        """Custom images with no hint produce no specs."""
+        compose = {
+            "services": {
+                "custom_app": {"image": "mycompany/custom-app:1.0"},
+                "web": {"image": "nginx:1.25"},
+            }
+        }
+        specs = generate_service_specs(compose, {"hosts": []})
+        assert len(specs) == 1
+        assert specs[0].daemon == "nginx"
+
+    def test_empty_compose(self):
+        """Empty compose falls through to topology."""
+        specs = generate_service_specs({}, {"hosts": ["web", "db"]})
+        daemon_names = {s.daemon for s in specs}
+        assert "nginx" in daemon_names
+        assert "mysqld" in daemon_names
+
+    def test_compose_env_vars_extracted(self):
+        """Environment variables from compose are passed to ServiceSpec."""
+        compose = {
+            "services": {
+                "db": {
+                    "image": "mysql:8.0",
+                    "environment": {"MYSQL_ROOT_PASSWORD": "secret"},
+                },
+            }
+        }
+        specs = generate_service_specs(compose, {"hosts": []})
+        assert len(specs) == 1
+        assert specs[0].env_vars.get("MYSQL_ROOT_PASSWORD") == "secret"
+
+    def test_compose_env_vars_list_form(self):
+        """Environment in list form (KEY=VALUE) is handled."""
+        compose = {
+            "services": {
+                "db": {
+                    "image": "mysql:8.0",
+                    "environment": ["MYSQL_ROOT_PASSWORD=secret", "MYSQL_DATABASE=app"],
+                },
+            }
+        }
+        specs = generate_service_specs(compose, {"hosts": []})
+        assert specs[0].env_vars["MYSQL_ROOT_PASSWORD"] == "secret"
+        assert specs[0].env_vars["MYSQL_DATABASE"] == "app"
+
+    def test_no_duplicate_daemons(self):
+        """If two compose services map to the same daemon, only one spec is produced."""
+        compose = {
+            "services": {
+                "siem": {"image": "rsyslog:latest"},
+                "firewall": {"image": "rsyslog:latest"},
+            }
+        }
+        specs = generate_service_specs(compose, {"hosts": []})
+        assert len(specs) == 1
+        assert specs[0].daemon == "rsyslogd"
+
+
+# ---------------------------------------------------------------------------
+# generate_service_specs() — topology fallback
+# ---------------------------------------------------------------------------
+
+
+class TestGenerateFromTopology:
+    """generate_service_specs() falls back to topology when compose is empty."""
+
+    def test_basic_topology_hosts(self):
+        topology = {
+            "hosts": ["attacker", "web", "db", "ldap", "siem", "files", "mail"],
+        }
+        specs = generate_service_specs({}, topology)
+
+        daemon_names = {s.daemon for s in specs}
+        assert "nginx" in daemon_names
+        assert "mysqld" in daemon_names
+        assert "slapd" in daemon_names
+        assert "rsyslogd" in daemon_names
+        assert "smbd" in daemon_names
+        assert "master" in daemon_names
+
+    def test_unknown_host_skipped(self):
+        topology = {"hosts": ["attacker", "custom_box"]}
+        specs = generate_service_specs({}, topology)
+        assert len(specs) == 0
+
+    def test_dict_hosts(self):
+        """Hosts as dicts with 'name' key."""
+        topology = {
+            "hosts": [
+                {"name": "web", "zone": "dmz"},
+                {"name": "db", "zone": "internal"},
+            ],
+        }
+        specs = generate_service_specs({}, topology)
+        daemon_names = {s.daemon for s in specs}
+        assert "nginx" in daemon_names
+        assert "mysqld" in daemon_names
+
+    def test_empty_topology(self):
+        specs = generate_service_specs({}, {})
+        assert specs == []
+
+
+# ---------------------------------------------------------------------------
+# _match_image_hint internals
+# ---------------------------------------------------------------------------
+
+
+class TestMatchImageHint:
+    """_match_image_hint matches Docker image strings to hint entries."""
+
+    def test_exact_match(self):
+        hint = _match_image_hint("nginx")
+        assert hint is not None
+        assert hint[0] == "nginx"
+
+    def test_tagged_image(self):
+        hint = _match_image_hint("mysql:8.0")
+        assert hint is not None
+        assert hint[0] == "mysqld"
+
+    def test_namespaced_image(self):
+        hint = _match_image_hint("osixia/openldap:1.5")
+        assert hint is not None
+        assert hint[0] == "slapd"
+
+    def test_basename_fallback(self):
+        """bitnami/redis:7 should match via basename 'redis'."""
+        hint = _match_image_hint("bitnami/redis:7")
+        assert hint is not None
+        assert hint[0] == "redis-server"
+
+    def test_unknown_image(self):
+        hint = _match_image_hint("mycompany/custom-service:v2")
+        assert hint is None
+
+    def test_empty_image(self):
+        hint = _match_image_hint("")
+        assert hint is None
+
+
+# ---------------------------------------------------------------------------
+# Environment integration: service lifecycle methods
+# ---------------------------------------------------------------------------
+
+
+class TestEnvironmentServiceLifecycle:
+    """RangeEnvironment service lifecycle methods."""
+
+    def test_start_snapshot_services_noop_in_docker_mode(self):
+        """_start_snapshot_services is a no-op when execution_mode != subprocess."""
+        from open_range.server.environment import RangeEnvironment
+
+        env = RangeEnvironment(docker_available=False)
+        # execution_mode defaults to "docker" when docker_available=False (mock)
+        snapshot = SnapshotSpec(
+            topology={"hosts": ["web"]},
+            services=[ServiceSpec(host="web", daemon="nginx", start_command="nginx &")],
+        )
+        # Should not raise or attempt to start anything
+        env._start_snapshot_services(snapshot)
+
+    @patch("subprocess.run")
+    def test_start_snapshot_services_subprocess_mode(self, mock_run):
+        """_start_snapshot_services starts declared services in subprocess mode."""
+        from open_range.server.environment import RangeEnvironment
+
+        env = RangeEnvironment(docker_available=False, execution_mode="subprocess")
+        snapshot = SnapshotSpec(
+            topology={"hosts": ["web"]},
+            services=[
+                ServiceSpec(
+                    host="web",
+                    daemon="nginx",
+                    init_commands=["mkdir -p /var/log/nginx"],
+                    start_command="nginx &",
+                    readiness=ReadinessCheck(type="tcp", port=80, timeout_s=0),
+                ),
+            ],
+        )
+        env._start_snapshot_services(snapshot)
+        # Should have called subprocess.run at least for init + start
+        assert mock_run.call_count >= 2
+
+    def test_start_services_empty_skips(self):
+        """When no services are declared, logs and skips provisioning."""
+        from open_range.server.environment import RangeEnvironment
+
+        env = RangeEnvironment(docker_available=False, execution_mode="subprocess")
+        snapshot = SnapshotSpec(
+            topology={"hosts": ["web", "db"]},
+            services=[],  # empty
+        )
+        # Should not raise — just logs and returns
+        env._start_snapshot_services(snapshot)
+
+    @patch("subprocess.run")
+    def test_stop_services_uses_snapshot_daemons(self, mock_run):
+        """_stop_services uses daemon names from snapshot.services."""
+        from open_range.server.environment import RangeEnvironment
+
+        env = RangeEnvironment(docker_available=False, execution_mode="subprocess")
+        env._snapshot = SnapshotSpec(
+            topology={"hosts": ["web"]},
+            services=[
+                ServiceSpec(host="web", daemon="nginx", start_command="nginx &"),
+                ServiceSpec(host="db", daemon="mysqld", start_command="mysqld &"),
+            ],
+        )
+        env._stop_services()
+
+        # Should have called pkill for each daemon (either individually or via bash -c)
+        all_call_strs = []
+        for call in mock_run.call_args_list:
+            args = call[0][0] if call[0] else call.kwargs.get("args", [])
+            all_call_strs.append(" ".join(str(a) for a in args))
+        combined = " ".join(all_call_strs)
+        assert "nginx" in combined
+        assert "mysqld" in combined
+
+    def test_stop_services_no_services_skips_pkill(self):
+        """_stop_services skips pkill when snapshot has no services."""
+        from open_range.server.environment import RangeEnvironment
+
+        env = RangeEnvironment(docker_available=False, execution_mode="subprocess")
+        env._snapshot = SnapshotSpec(topology={"hosts": ["web"]})
+        # Should not raise — just skips pkill since no service specs
+        env._stop_services()
+
+    def test_stop_services_no_snapshot(self):
+        """_stop_services handles None snapshot gracefully."""
+        from open_range.server.environment import RangeEnvironment
+
+        env = RangeEnvironment(docker_available=False, execution_mode="subprocess")
+        env._snapshot = None
+        # Should not raise
+        env._stop_services()
+
+    def test_probe_readiness_tcp_unreachable(self):
+        """TCP probe returns False for unreachable port."""
+        from open_range.server.environment import RangeEnvironment
+
+        check = ReadinessCheck(type="tcp", port=19999)
+        assert RangeEnvironment._probe_readiness(check) is False
+
+    def test_probe_readiness_command_success(self):
+        """Command probe returns True for 'true' command."""
+        from open_range.server.environment import RangeEnvironment
+
+        check = ReadinessCheck(type="command", command="true")
+        assert RangeEnvironment._probe_readiness(check) is True
+
+    def test_probe_readiness_command_failure(self):
+        """Command probe returns False for 'false' command."""
+        from open_range.server.environment import RangeEnvironment
+
+        check = ReadinessCheck(type="command", command="false")
+        assert RangeEnvironment._probe_readiness(check) is False
+
+    def test_reset_calls_service_lifecycle(self):
+        """reset() calls _stop_services and _start_snapshot_services."""
+        from open_range.server.environment import RangeEnvironment
+
+        env = RangeEnvironment(docker_available=False)
+        stop_called = []
+        start_called = []
+
+        env._stop_services = lambda: stop_called.append(True)  # type: ignore
+        env._start_snapshot_services = lambda s: start_called.append(s)  # type: ignore
+
+        snapshot = SnapshotSpec(
+            topology={"hosts": ["attacker", "web"]},
+            task=TaskSpec(red_briefing="Test.", blue_briefing="Test."),
+        )
+        env.reset(snapshot=snapshot)
+        assert len(stop_called) == 1
+        assert len(start_called) == 1
+
+
+# ---------------------------------------------------------------------------
+# Renderer generates services in snapshot
+# ---------------------------------------------------------------------------
+
+
+class TestRendererServiceGeneration:
+    """SnapshotRenderer._build_service_specs() populates spec.services."""
+
+    def test_renderer_populates_services_from_topology(self):
+        from open_range.builder.renderer import SnapshotRenderer
+
+        renderer = SnapshotRenderer()
+        spec = SnapshotSpec(
+            topology={
+                "hosts": ["web", "db", "ldap"],
+                "zones": {"dmz": ["web"], "internal": ["db", "ldap"]},
+                "users": [],
+                "firewall_rules": [],
+            },
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            renderer.render(spec, Path(tmpdir) / "out")
+
+        # After rendering, services should be populated
+        assert len(spec.services) >= 2
+        daemon_names = {s.daemon for s in spec.services}
+        assert "nginx" in daemon_names
+        assert "mysqld" in daemon_names
+
+    def test_renderer_skips_if_services_already_present(self):
+        from open_range.builder.renderer import SnapshotRenderer
+
+        renderer = SnapshotRenderer()
+        existing_svc = ServiceSpec(host="web", daemon="nginx", start_command="nginx &")
+        spec = SnapshotSpec(
+            topology={
+                "hosts": ["web", "db"],
+                "zones": {"dmz": ["web"], "internal": ["db"]},
+                "users": [],
+                "firewall_rules": [],
+            },
+            services=[existing_svc],
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            renderer.render(spec, Path(tmpdir) / "out")
+
+        # Should not have overwritten — still just the one we provided
+        assert len(spec.services) == 1
+        assert spec.services[0].daemon == "nginx"
+
+
+# ---------------------------------------------------------------------------
+# Hint table coverage
+# ---------------------------------------------------------------------------
+
+
+class TestHintTableCoverage:
+    """All image hints produce valid ServiceSpec entries."""
+
+    @pytest.mark.parametrize("image_key", list(_IMAGE_SERVICE_HINTS.keys()))
+    def test_hint_produces_valid_spec(self, image_key):
+        """Each entry in the hint table produces a valid ServiceSpec."""
+        compose = {"services": {"svc": {"image": image_key}}}
+        specs = generate_service_specs(compose, {"hosts": []})
+        assert len(specs) == 1
+        svc = specs[0]
+        assert svc.daemon
+        assert svc.start_command
+        assert isinstance(svc.readiness, ReadinessCheck)
+
+    @pytest.mark.parametrize("host_name", list(_HOST_NAME_HINTS.keys()))
+    def test_host_hint_produces_valid_spec(self, host_name):
+        """Each entry in the host-name hint table produces a valid ServiceSpec."""
+        specs = generate_service_specs({}, {"hosts": [host_name]})
+        assert len(specs) >= 1
+        svc = specs[0]
+        assert svc.daemon
+        assert svc.start_command

tests/test_validator.py

@@ -426,12 +426,11 @@ async def test_patchability_skips_prose_remediation(mock_containers):
 
     result = await PatchabilityCheck().check(spec, mock_containers)
     assert result.passed is False
-    assert "no vulns had testable remediation" in result.error
-    # Verify it was recorded as skipped
+    # Verify it was recorded as a failure (not silently skipped)
     vuln_results = result.details["vuln_results"]
     assert len(vuln_results) == 1
-    assert "skipped" in vuln_results[0]
-    assert "not executable" in vuln_results[0]["skipped"]
+    assert vuln_results[0]["passed"] is False
+    assert "not executable" in vuln_results[0]["reason"]
 
 
 @pytest.mark.asyncio