feat: implement RL environment server with training infrastructure and Modal integration

01_ARCHITECTURE.md

@@ -22,64 +22,56 @@ Editable source: `assets/architecture_diagram.mmd`
 
 ```mermaid
 flowchart TB
-    %% =========================
-    %% Offline Build Layer
-    %% =========================
-    subgraph A[Offline Scenario Factory]
-        A1[Policy Graph Generator\nroles, users, tenants, ownership, route intent]
-        A2[App Template Library\nFastAPI, Express, Django MVP templates]
-        A3[Bug Injector\nmissing guard, IDOR, tenant leak, role confusion, query omission]
-        A4[Scenario Compiler\nmaterializes app + DB + public tests + hidden invariants]
-        A5[Split Manager\ntrain seeds, validation seeds, hidden held-out seeds]
-        A1 --> A4
-        A2 --> A4
-        A3 --> A4
-        A5 --> A4
+    subgraph A[Scenario + Curriculum Factory]
+        A1[Policy Graph Generator\nroles, users, tenants, ownership]
+        A2[Curriculum Controller\nmastery, weak spots, difficulty tier]
+        A3[Bounded Adversarial Designer\nsafe local scenario targets]
+        A4[Template Renderer\nFastAPI routes, services, auth helpers]
+        A5[A01 Bug Mutator\nIDOR, tenant, role, public-route traps]
+        A6[ScenarioSpec + Oracle\nvisible hints + hidden policy tuples]
+        A1 --> A3
+        A2 --> A3
+        A3 --> A4 --> A5 --> A6
     end
 
-    %% =========================
-    %% OpenEnv Runtime
-    %% =========================
     subgraph B[CyberSecurity_OWASP OpenEnv Server]
-        B1[reset\(\)\nselect scenario + start sandbox]
-        B2[Sandbox App Runtime\nlocal app, DB fixture, logs, route map]
-        B3[Tool API exposed through step\(action\)\nReadFile, ListRoutes, SendLocalRequest, RunTests, ApplyPatch, SubmitFix]
-        B4[State Store\nepisode_id, step_count, scenario_id, patch diff, test history]
-        B5[Deterministic Reward Engine\npolicy tests + hidden tests + regression tests + penalties]
-        B6[state\(\)\nstructured metadata for debugging/eval]
-        B1 --> B2
-        B2 --> B3
-        B3 --> B4
-        B4 --> B5
-        B4 --> B6
+        B1[reset\(seed, difficulty\)\nselect curriculum profile]
+        B2[Episode State Store\nphase, history, metrics, weakness, patch diff]
+        B3[Typed Action Tools\ninspect, request, patch, visible tests]
+        B4[Ephemeral App Sandbox\ncode workspace + fixtures + local API model]
+        B5[Multi-layer Verifier\nvisible, hidden, oracle, regression]
+        B6[Deterministic Reward Engine\nstable components + penalties]
+        B7[Episode Artifact Logger\nJSONL transcript + verifier + diff]
+        B8[state\(\)\nstructured metadata for debugging/eval]
+        B1 --> B2 --> B3
+        B3 <--> B4
+        B4 --> B5 --> B6 --> B2
+        B2 --> B7 --> A2
+        B2 --> B8
     end
 
-    %% =========================
-    %% Agent + Training
-    %% =========================
     subgraph C[Single LLM Agent]
         C1[Observation Parser]
-        C2[Planner\npolicy reasoning + patch strategy]
-        C3[Action Generator\nchooses next OpenEnv action]
+        C2[AuthZ + Code Reasoning]
+        C3[Discover → Diagnose → Patch → Test\none JSON action]
         C1 --> C2 --> C3
     end
 
     subgraph D[Training + Evaluation]
-        D1[Rollout Loop\nreset → step* → final reward]
-        D2[GRPO / TRL / Unsloth Training]
-        D3[Trackio Metrics\nreward curves, pass rates, patch size, steps]
-        D4[Held-out Eval Suite\nunseen templates, seeds, names, route structures]
-        D5[Demo Artifacts\nbefore/after traces, mini-blog, 2-minute video]
-        D1 --> D2 --> D3
-        D3 --> D4 --> D5
+        D1[Parallel Rollout Loop\nreset → step* → terminal reward]
+        D2[TRL GRPO + LoRA]
+        D3[Trackio Metrics\nreward curves, pass rates, failure modes]
+        D4[Held-out Family Eval\nbase vs trained model]
+        D5[Demo Artifacts\nbefore/after traces + JSONL]
+        D1 --> D2 --> D3 --> D4 --> D5
     end
 
-    A4 --> B1
+    A6 --> B1
     C3 -->|typed action| B3
     B3 -->|observation + reward + done| C1
-    B5 --> D1
+    B6 --> D1
     D2 --> C1
-    B5 --> D4
+    B6 --> D4
 ```
 
 ## 3. Component responsibilities
@@ -113,6 +105,13 @@ The scenario compiler is the main anti-overfitting mechanism. It should vary:
 - visible test coverage;
 - hidden invariant seeds.
 
+The runtime now treats curriculum and adversarial targeting as first-class scenario inputs:
+
+- `CurriculumController` tracks target weakness mastery, recent reward trend, failure counts, and difficulty tier.
+- `BoundedAdversarialDesigner` chooses safe synthetic lab targets such as same-role cross-object access, cross-tenant boundaries, public-route overlocking, alternate-service reachability, and visible-test-only traps.
+- `ScenarioFactory` combines the policy graph, curriculum profile, adversarial target, renderer, and hidden oracle metadata into one deterministic scenario spec.
+- Hidden-eval episodes hold out scenario families, not only seeds, by marking evaluation-only scenario-family metadata in state rather than observations.
+
 ### 3.2 Policy Graph Generator
 
 The policy graph is the ground truth for intended behavior.
@@ -222,16 +221,19 @@ Observations should be compact and structured.
 ```python
 @dataclass
 class CyberSecurityOWASPObservation(Observation):
+    phase: Literal["discover", "patch", "done"]
     message: str
-    visible_policy_summary: str
-    route_summary: list[dict]
-    last_action_result: dict
-    public_test_summary: dict
-    patch_summary: dict
+    task_brief: str
+    visible_policy_hint: dict
+    workspace_summary: dict
+    available_actions: list[str]
+    last_tool_result: str
+    visible_test_result: str | None = None
+    reward_breakdown: dict[str, float] = field(default_factory=dict)
     done_reason: str | None = None
 ```
 
-Do not expose hidden test bodies, hidden expected outputs, or seed-specific solution hints.
+The policy hint is deliberately partial. It may include product rules, fixture aliases, route summaries, and public-route intent, but it must not expose the hidden oracle matrix, hidden test bodies, injected bug labels, or held-out family labels.
 
 ### 3.7 State schema
 
@@ -241,17 +243,18 @@ State should support debugging and training analytics.
 @dataclass
 class CyberSecurityOWASPState(State):
     episode_id: str
-    scenario_id: str
-    split: Literal["train", "validation", "heldout"]
+    task_id: str
+    split: Literal["train", "validation", "hidden_eval"]
     step_count: int = 0
-    max_steps: int = 30
+    max_steps: int = 40
+    difficulty_tier: str = "warmup"
     scenario_family: str = ""
-    app_template: str = ""
-    files_touched: list[str] = field(default_factory=list)
-    public_tests_passed: int = 0
-    public_tests_total: int = 0
-    hidden_tests_passed: int = 0
-    hidden_tests_total: int = 0
+    template_id: str = "fastapi_basic"
+    target_weakness: str = ""
+    curriculum_snapshot: dict = field(default_factory=dict)
+    verification_summary: dict = field(default_factory=dict)
+    patch_diff: str = ""
+    episode_artifact_path: str | None = None
     accumulated_reward: float = 0.0
 ```
 
@@ -259,9 +262,10 @@ class CyberSecurityOWASPState(State):
 
 ```text
 1. reset()
-   - sample train/validation scenario seed
-   - compile app from policy graph + template + injected bug
-   - start local sandbox app and DB fixture
+   - curriculum selects difficulty tier and target weakness
+   - bounded adversarial designer chooses a safe local scenario target
+   - scenario factory compiles app from policy graph + template + injected bug
+   - initialize ephemeral app sandbox and fixture state
    - return initial observation
 
 2. agent loop
@@ -275,52 +279,58 @@ class CyberSecurityOWASPState(State):
    - freeze patch
    - run public tests
    - run hidden authorization invariants
-   - run regression tests
+   - run policy-oracle matrix
+   - run regression and public-route preservation tests
    - compute deterministic reward
    - return final observation, reward, done=True
 
 4. logging
-   - record scenario_id, action trace, patch diff, reward components
+   - append JSONL artifact with scenario metadata, action trace, observations, patch diff, verifier result, and reward components
+   - feed terminal success/failure back into curriculum mastery tracking
    - send metrics to Trackio during training/eval
 ```
 
 ## 5. Reward design
 
-The reward should be deterministic, decomposed, and resistant to reward hacking.
+The reward should be deterministic, decomposed, and resistant to reward hacking. The maximum terminal reward remains **15.0** and high reward requires deterministic verifier success, not explanation quality.
 
-Recommended reward formula:
+Stable reward keys:
 
-```text
-R = 0.35 * public_policy_tests
-  + 0.30 * hidden_authz_invariants
-  + 0.15 * regression_preservation
-  + 0.10 * evidence_quality
-  + 0.05 * patch_minimality
-  + 0.05 * efficiency
-  - penalties
+```python
+{
+    "discovery": 0.0,
+    "security": 0.0,
+    "regression": 0.0,
+    "public_routes": 0.0,
+    "patch_quality": 0.0,
+    "visible_tests": 0.0,
+    "safety": 0.0,
+    "anti_cheat": 0.0,
+    "total": 0.0,
+}
 ```
 
 ### Reward components
 
-| Component | Weight | What it rewards |
-|---|---:|---|
-| Public policy tests | 0.35 | Agent fixes known failing behavior. |
-| Hidden authz invariants | 0.30 | Patch generalizes beyond visible tests. |
-| Regression preservation | 0.15 | Valid user flows and intended public routes still work. |
-| Evidence quality | 0.10 | Agent gathered relevant policy/test/file evidence before patching. |
-| Patch minimality | 0.05 | Small focused patches instead of broad rewrites. |
-| Efficiency | 0.05 | Fewer wasted steps and repeated actions. |
+| Component | Purpose |
+|---|---|
+| `discovery` | Valid local evidence and correct violated policy rule. |
+| `security` | Hidden exploit blocking plus policy-oracle matrix pass. |
+| `regression` | Legitimate owner/admin/support flows still work. |
+| `public_routes` | Intentionally public routes remain public. |
+| `patch_quality` | Localized policy-aligned patch and efficient phase order. |
+| `visible_tests` | Visible tests pass and app still boots. |
+| `safety` | Penalizes invalid action patterns, unsafe targets, timeouts, and deny-all behavior. |
+| `anti_cheat` | Penalizes hidden-file probing, hardcoded fixture IDs, and test/oracle tampering. |
 
 ### Penalties
 
 | Penalty | Trigger |
 |---|---|
-| `-0.25` | Breaks public route intentionally marked public. |
-| `-0.25` | Deletes tests, policy file, or route instead of fixing authorization. |
-| `-0.20` | Hardcodes seed-specific IDs, users, tenants, or hidden assumptions. |
-| `-0.15` | Over-broad denial that blocks legitimate authorized users. |
-| `-0.10` | Patch exceeds file or diff-size budget. |
-| `-1.00` | Attempts external network access, credential extraction, persistence, or unsafe behavior. |
+| public route penalty | Breaks a route intentionally marked public. |
+| anti-cheat penalty | Deletes or probes tests, hidden files, reward code, oracle data, or host paths. |
+| hardcoding penalty | Hardcodes seed-specific IDs, users, tenants, or hidden assumptions. |
+| safety penalty | Over-broad denial, malformed/invalid actions, repeated failed actions, or external target attempts. |
 
 The LLM judge, if used at all, should only annotate trace quality for analysis. It must not decide security-critical reward.
 
@@ -488,3 +498,4 @@ Expected endpoints:
 | OpenEnv deployment docs | Informs HF Spaces deployment, endpoints, Docker workflow, and installable client package. | 8.5/10 |
 | Hackathon judging criteria | Informs demo priorities: innovation, storytelling, reward improvement, and training pipeline. | 9/10 |
 | TRL/OpenEnv training example | Informs rollout function, decomposed reward functions, and Trackio logging pattern. | 8/10 |
+| Kube SRE Gym README | Informs the closed-loop pattern: adversarial scenario design, curriculum mastery tracking, real tool interaction, verification, and artifact-driven storytelling. | 8/10 |

README.md

@@ -21,7 +21,7 @@ tags:
 inspect generated app + policy -> discover authorization bug -> submit finding -> patch code -> preserve intended behavior
 ```
 
-The current implementation includes a functional MVP scenario: an invoices FastAPI-style app with one injected OWASP A01 BOLA/IDOR defect, visible tests, hidden deterministic verifier checks, anti-cheat safeguards, and decomposed reward.
+The current implementation includes a functional closed-loop MVP scenario: an invoices FastAPI-style app with one injected OWASP A01 BOLA/IDOR defect, curriculum-aware scenario selection, bounded adversarial targeting, an ephemeral app sandbox, multi-layer deterministic verifier checks, anti-cheat safeguards, JSONL episode artifacts, and decomposed reward.
 
 ## Diagrams
 
@@ -98,27 +98,43 @@ Terminal reward uses stable components:
 }
 ```
 
-The verifier rewards blocking the hidden exploit while preserving legitimate owner/admin behavior and intentionally public routes. It penalizes deny-all fixes, hardcoded IDs, hidden file probes, external URL attempts, and test/fixture tampering.
+The verifier rewards blocking the hidden exploit while preserving legitimate owner/admin behavior and intentionally public routes. Terminal scoring requires visible checks, hidden authorization checks, a policy-oracle matrix, regression checks, public-route preservation, and patch-quality checks. It penalizes deny-all fixes, hardcoded IDs, repeated/invalid action patterns, hidden file probes, external URL attempts, and test/fixture tampering.
 
 ## Scenario Generation
 
-`reset(seed)` compiles a fresh isolated workspace under a temp directory. The MVP compiler generates:
+`reset(seed)` asks the `CurriculumController` for a difficulty tier and target weakness, then `ScenarioFactory` uses a bounded adversarial designer to compile a fresh isolated workspace under a temp directory. The MVP compiler generates:
 
 - invoices domain policy graph;
+- bounded adversarial target metadata such as same-role cross-object access, cross-tenant access, public-route overlocking traps, alternate route/service reachability, or visible-test-only edge cases;
 - randomized users, tenants, invoices, and IDs;
 - generated app files under `app/`;
 - visible tests under `tests/test_visible.py`;
-- hidden facts kept only in state for deterministic verification.
+- hidden facts, oracle tuples, scenario family metadata, and verifier targets kept out of observations.
 
 Additional domains and bug families are scaffolded for extension.
 
+## Runtime Components
+
+The OpenEnv runtime is split into small server modules:
+
+- `server/curriculum.py` tracks mastery, weak spots, reward trend, and difficulty tier.
+- `server/adversarial_designer.py` chooses safe synthetic scenario targets from tracked weaknesses.
+- `server/scenario_factory.py` compiles the generated app, visible hints, hidden facts, scenario family, and template metadata.
+- `server/app_sandbox.py` handles editable workspace reads, patches, local requests, and OpenAPI summaries.
+- `server/action_tools.py` dispatches typed tools through the sandbox.
+- `server/authz_oracle.py` builds the hidden allowed/denied user-resource-action matrix.
+- `server/verifier.py` aggregates visible tests, hidden tests, oracle matrix, regression/public-route checks, and patch quality.
+- `server/episode_logger.py` appends JSONL rollouts under `outputs/rollouts/`.
+
+The agent sees partial observations only: product rules, fixture aliases, route summaries, visible test results, and action errors. Hidden tests, oracle tuples, injected bug labels, and held-out scenario-family labels stay internal.
+
 ## Testing
 
 ```bash
 uv run --extra dev pytest
 ```
 
-The suite covers model serialization, reset/step/state behavior, seed reproducibility, invalid actions, reward outcomes, anti-cheat checks, and scripted rollout policies.
+The suite covers model serialization, reset/step/state behavior, seed reproducibility, invalid actions, reward outcomes, anti-cheat checks, scripted rollout policies, curriculum selection, adversarial targeting, held-out scenario families, oracle checks, verifier aggregation, and episode artifact logging.
 
 ## Training Scaffold
 

assets/architecture_diagram.mmd

@@ -1,46 +1,51 @@
-flowchart LR
-    subgraph Factory["Scenario Factory"]
-        Policy["Policy graph\nusers, roles, tenants, ownership"]
-        Templates["FastAPI template renderer\nroutes, services, auth helpers"]
-        Mutator["A01 bug mutator\none injected authorization defect"]
-        Fixtures["Fixture generator\nvisible tests + hidden facts"]
-        Compiler["Scenario compiler\nseeded workspace"]
-        Policy --> Compiler
-        Templates --> Compiler
+flowchart TB
+    subgraph Factory["Scenario + Curriculum Factory"]
+        Policy["Policy graph generator\nusers, roles, tenants, ownership"]
+        Curriculum["Curriculum controller\nmastery, weak spots, difficulty tier"]
+        Designer["Bounded adversarial designer\nsafe local scenario targets"]
+        Templates["Template renderer\nFastAPI routes, services, auth helpers"]
+        Mutator["A01 bug mutator\nIDOR, tenant, role, public-route traps"]
+        Compiler["ScenarioSpec + oracle\nvisible hints + hidden policy tuples"]
+        Policy --> Designer
+        Curriculum --> Designer
+        Designer --> Templates
+        Templates --> Mutator
         Mutator --> Compiler
-        Fixtures --> Compiler
     end
 
     subgraph Runtime["CyberSecurity_OWASP OpenEnv Runtime"]
-        Reset["reset(seed)\ncompile fresh scenario"]
-        Env["Environment state\nphase, history, metrics, hidden facts"]
-        Tools["Typed step(action) tools\ninspect, read, request, patch, test, submit"]
-        Sandbox["Generated local app workspace\neditable app files only"]
-        Verifier["Deterministic verifier\nsecurity + regression + public routes"]
-        Reward["Reward engine\nstable component breakdown"]
-        App["FastAPI OpenEnv server\n/ws, /reset, /step, /state"]
-        Reset --> Env
-        Env --> Tools
+        Reset["reset(seed, difficulty)\nselect curriculum profile"]
+        State["Episode state store\nphase, history, metrics, weakness, patch diff"]
+        Tools["Typed action tools\ninspect, request, patch, visible tests"]
+        Sandbox["Ephemeral app sandbox\ncode workspace + fixture DB + local API model"]
+        Verifier["Multi-layer verifier\nvisible, hidden, oracle, regression"]
+        Reward["Deterministic reward engine\ncomponents + penalties"]
+        Logger["Episode artifact logger\nJSONL transcript + verifier + diff"]
+        App["FastAPI / WebSocket server\n/ws, /reset, /step, /state, /web"]
+        Reset --> State
+        State --> Tools
         Tools <--> Sandbox
-        Tools --> Verifier
+        Sandbox --> Verifier
         Verifier --> Reward
-        Reward --> Env
-        Env --> App
+        Reward --> State
+        State --> Logger
+        Logger --> Curriculum
+        State --> App
     end
 
     subgraph Agent["Single LLM Agent"]
         Obs["Observation parser"]
-        Reason["Policy and code reasoning"]
-        Act["One JSON action"]
+        Reason["AuthZ + code reasoning"]
+        Act["Discover -> Diagnose -> Patch -> Test\none JSON action at a time"]
         Obs --> Reason --> Act
     end
 
-    subgraph Ops["Training, Evaluation, Demo"]
-        Rollout["Rollout loop\nreset -> step* -> terminal reward"]
-        GRPO["TRL GRPO / LoRA training"]
-        Trackio["Trackio metrics\nreward and pass rates"]
-        Eval["Held-out evaluation\nunseen seeds/layouts/domains"]
-        Artifacts["Rollout artifacts\nbefore/after traces"]
+    subgraph Ops["Training / Evaluation / Demo"]
+        Rollout["Parallel rollout loop\nreset -> step* -> terminal reward"]
+        GRPO["TRL GRPO + LoRA"]
+        Trackio["Trackio reward curves\npass rates and failure modes"]
+        Eval["Held-out family eval\nbase vs trained model"]
+        Artifacts["Demo artifacts\nbefore/after traces + JSONL"]
         Rollout --> GRPO --> Trackio --> Eval --> Artifacts
     end
 

models.py

@@ -56,8 +56,12 @@ class CyberSecurityOWASPState(State):
     seed: int = 0
     split: CyberSecurityOWASPSplit = "train"
     difficulty: int = 0
+    difficulty_tier: str = "warmup"
     domain: str = ""
     bug_family: str = ""
+    scenario_family: str = ""
+    template_id: str = "fastapi_basic"
+    target_weakness: str = "same_role_cross_object"
     phase: CyberSecurityOWASPPhase = "discover"
     max_steps: int = 40
     done: bool = False
@@ -71,6 +75,11 @@ class CyberSecurityOWASPState(State):
     reward_history: list[dict[str, float]] = Field(default_factory=list)
     visible_facts: dict[str, Any] = Field(default_factory=dict)
     hidden_facts: dict[str, Any] = Field(default_factory=dict)
+    curriculum_snapshot: dict[str, Any] = Field(default_factory=dict)
+    verification_summary: dict[str, Any] = Field(default_factory=dict)
+    patch_diff: str = ""
+    episode_artifact_path: str | None = None
+    observation_history: list[dict[str, Any]] = Field(default_factory=list)
     metrics: dict[str, Any] = Field(default_factory=dict)
     anti_cheat_flags: list[str] = Field(default_factory=list)
 

pyproject.toml

@@ -47,3 +47,14 @@ server = "CyberSecurity_OWASP.server.app:main"
 include-package-data = true
 packages = ["CyberSecurity_OWASP", "CyberSecurity_OWASP.server"]
 package-dir = { "CyberSecurity_OWASP" = ".", "CyberSecurity_OWASP.server" = "server" }
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+norecursedirs = [
+    ".git",
+    ".venv",
+    "__pycache__",
+    "codex_tmp_tf",
+    "outputs",
+    "openenv_CyberSecurity_OWASP.egg-info",
+]

rewards.py

@@ -41,21 +41,28 @@ def compute_reward(
         reward["visible_tests"] = 1.0 if visible.get("passed") else 0.0
     elif action.tool_name == "submit_fix":
         security = verifier_result.get("security", {})
+        oracle = verifier_result.get("oracle_matrix", {})
         regression = verifier_result.get("regression", {})
         public_routes = verifier_result.get("public_routes", {})
         quality = verifier_result.get("patch_quality", {})
         visible = verifier_result.get("visible", {})
-        reward["security"] = 5.0 if security.get("passed") else -6.0
+        security_passed = bool(security.get("passed") and oracle.get("passed", True))
+        reward["security"] = 5.0 if security_passed else -6.0
         reward["regression"] = 3.0 if regression.get("passed") else -5.0
         reward["public_routes"] = 1.0 if public_routes.get("passed") else -3.0
         reward["patch_quality"] = 2.0 if quality.get("passed") else 0.0
+        if quality.get("passed") and _followed_repair_phase_order(state):
+            reward["patch_quality"] += 0.5
         reward["visible_tests"] = 1.0 if visible.get("passed") else -6.0
         if quality.get("hardcoded"):
             reward["anti_cheat"] -= 4.0
         if quality.get("deny_all"):
             reward["safety"] -= 5.0
 
-    flags = verifier_result.get("anti_cheat_flags", []) or []
+    flags = sorted(
+        set(verifier_result.get("anti_cheat_flags", []) or [])
+        | set(state.anti_cheat_flags or [])
+    )
     if flags:
         if "external_network_attempt" in flags:
             reward["safety"] -= 8.0
@@ -64,6 +71,21 @@ def compute_reward(
         if "hardcoded_hidden_identifier" in flags:
             reward["anti_cheat"] -= 4.0
 
+    if verifier_result.get("invalid_action"):
+        reward["safety"] -= 0.5
+    if verifier_result.get("repeated_action"):
+        reward["safety"] -= 0.2
+
     total = sum(value for key, value in reward.items() if key != "total")
     reward["total"] = min(15.0, total) if total > 0 else total
     return reward
+
+
+def _followed_repair_phase_order(state: CyberSecurityOWASPState) -> bool:
+    tools = [item.get("tool_name") for item in state.action_history]
+    required = ["submit_finding", "patch_file", "run_visible_tests", "submit_fix"]
+    cursor = 0
+    for tool in tools:
+        if cursor < len(required) and tool == required[cursor]:
+            cursor += 1
+    return cursor == len(required)

scenario_compiler.py

@@ -1,66 +1,26 @@
-"""Deterministic scenario compiler for CyberSecurity_OWASP."""
+"""Compatibility facade for deterministic scenario compilation."""
 
 from __future__ import annotations
 
-import os
-import tempfile
-from pathlib import Path
 from typing import Any
-from uuid import uuid4
 
 try:
-    from .fixture_generator import visible_workspace_summary
-    from .policy_graph import build_invoice_policy
-    from .template_renderer import render_fastapi_basic
+    from .server.scenario_factory import ScenarioFactory
 except ImportError:  # pragma: no cover
-    from fixture_generator import visible_workspace_summary
-    from policy_graph import build_invoice_policy
-    from template_renderer import render_fastapi_basic
-
-
-def _make_workspace(prefix: str) -> Path:
-    root = Path(os.getenv("CYBERSECURITY_OWASP_WORKSPACE_ROOT", tempfile.gettempdir()))
-    root.mkdir(parents=True, exist_ok=True)
-    for _ in range(100):
-        workspace = root / f"{prefix}{uuid4().hex[:12]}"
-        try:
-            workspace.mkdir()
-        except FileExistsError:
-            continue
-        return workspace
-    raise RuntimeError("Unable to create isolated scenario workspace")
-
-
-def compile_scenario(seed: int, split: str = "train", difficulty: int = 0) -> dict[str, Any]:
-    """Compile one isolated MVP authorization-repair scenario."""
-
-    compiled = build_invoice_policy(seed)
-    workspace = _make_workspace(prefix=f"cybersecurity_owasp_{split}_{seed}_")
-    editable_files = render_fastapi_basic(workspace, compiled.public_hint, compiled.hidden_facts)
-    task_id = f"{split}-invoices-bola-{seed}"
-    hidden = dict(compiled.hidden_facts)
-    hidden.update(
-        {
-            "workspace": str(workspace),
-            "editable_files": editable_files,
-            "initial_file_hashes": {
-                path: (workspace / path).read_text(encoding="utf-8")
-                for path in editable_files
-            },
-        }
+    from server.scenario_factory import ScenarioFactory
+
+
+def compile_scenario(
+    seed: int,
+    split: str = "train",
+    difficulty: int = 0,
+    curriculum_profile: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Compile one isolated authorization-repair scenario."""
+
+    return ScenarioFactory().compile_scenario(
+        seed,
+        split=split,
+        difficulty=difficulty,
+        curriculum_profile=curriculum_profile,
     )
-    return {
-        "task_id": task_id,
-        "workspace": workspace,
-        "domain": "invoices",
-        "bug_family": "bola_idor",
-        "difficulty": difficulty,
-        "task_brief": (
-            "Inspect the generated invoices app and policy. Find the broken "
-            "authorization behavior, submit a finding with local evidence, patch "
-            "the app, preserve intended owner/admin/public behavior, then submit."
-        ),
-        "public_hint": compiled.public_hint,
-        "workspace_summary": visible_workspace_summary(editable_files, compiled.public_hint),
-        "hidden_facts": hidden,
-    }

scripts/modal_ephemeral_train.py

@@ -35,6 +35,8 @@ image = (
         ignore=[
             ".git",
             ".venv",
+            ".env",
+            ".env.*",
             "__pycache__",
             ".pytest_cache",
             "outputs",

scripts/modal_train_grpo.py

@@ -135,6 +135,8 @@ def _training_image() -> modal.Image:
             ignore=[
                 ".git",
                 ".venv",
+                ".env",
+                ".env.*",
                 "__pycache__",
                 ".pytest_cache",
                 "outputs",

scripts/track_pytest.py

@@ -10,6 +10,7 @@ from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(PROJECT_ROOT))
+sys.path.insert(0, str(PROJECT_ROOT.parent))
 
 from training.trackio_utils import build_run_name, get_git_sha, log_trackio_metrics, trackio_run
 

server/CyberSecurity_OWASP_environment.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 import json
 import shutil
-from pathlib import Path
 from typing import Any
 from uuid import uuid4
 
@@ -16,16 +15,20 @@ try:
         CyberSecurityOWASPObservation,
         CyberSecurityOWASPState,
     )
-    from ..scenario_compiler import compile_scenario
-    from ..safety import is_local_route
-    from ..validators import detect_cheating, is_path_allowed, simulate_request
+    from ..validators import detect_cheating
+    from .action_tools import ActionTools
+    from .curriculum import CurriculumController
+    from .episode_logger import EpisodeArtifactLogger
     from .reward_engine import evaluate_action
+    from .scenario_factory import ScenarioFactory
 except ImportError:  # pragma: no cover
     from models import CyberSecurityOWASPAction, CyberSecurityOWASPObservation, CyberSecurityOWASPState
-    from scenario_compiler import compile_scenario
-    from safety import is_local_route
-    from validators import detect_cheating, is_path_allowed, simulate_request
+    from validators import detect_cheating
+    from server.action_tools import ActionTools
+    from server.curriculum import CurriculumController
+    from server.episode_logger import EpisodeArtifactLogger
     from server.reward_engine import evaluate_action
+    from server.scenario_factory import ScenarioFactory
 
 
 ALLOWED_TOOLS = {
@@ -67,6 +70,9 @@ class CybersecurityOwaspEnvironment(
         self._visible_policy_hint: dict[str, Any] = {}
         self._workspace_summary: dict[str, Any] = {}
         self._last_done_observation: CyberSecurityOWASPObservation | None = None
+        self._curriculum = CurriculumController()
+        self._scenario_factory = ScenarioFactory()
+        self._episode_logger = EpisodeArtifactLogger()
 
     def reset(
         self,
@@ -78,15 +84,29 @@ class CybersecurityOwaspEnvironment(
     ) -> CyberSecurityOWASPObservation:
         self.close()
         actual_seed = int(seed if seed is not None else 0)
-        scenario = compile_scenario(actual_seed, split=split, difficulty=difficulty)
+        curriculum_profile = self._curriculum.select_profile(
+            seed=actual_seed,
+            split=split,
+            requested_difficulty=difficulty,
+        )
+        scenario = self._scenario_factory.compile_scenario(
+            actual_seed,
+            split=split,
+            difficulty=difficulty,
+            curriculum_profile=curriculum_profile,
+        )
         self._state = CyberSecurityOWASPState(
             episode_id=episode_id or str(uuid4()),
             task_id=scenario["task_id"],
             seed=actual_seed,
             split=split,
-            difficulty=difficulty,
+            difficulty=scenario["difficulty"],
+            difficulty_tier=scenario["difficulty_tier"],
             domain=scenario["domain"],
             bug_family=scenario["bug_family"],
+            scenario_family=scenario["scenario_family"],
+            template_id=scenario["template_id"],
+            target_weakness=scenario["target_weakness"],
             phase="discover",
             step_count=0,
             max_steps=40,
@@ -94,6 +114,7 @@ class CybersecurityOwaspEnvironment(
             success=False,
             visible_facts={"workspace_summary": scenario["workspace_summary"]},
             hidden_facts=scenario["hidden_facts"],
+            curriculum_snapshot=scenario["curriculum_snapshot"],
             metrics={"reset_count": 1},
         )
         self._task_brief = scenario["task_brief"]
@@ -124,7 +145,12 @@ class CybersecurityOwaspEnvironment(
         )
 
         if action.tool_name not in ALLOWED_TOOLS[self._state.phase]:
-            verifier, reward = evaluate_action(self._state, action, anti_cheat_flags)
+            verifier, reward = evaluate_action(
+                self._state,
+                action,
+                anti_cheat_flags,
+                invalid_action=True,
+            )
             return self._finish_step(
                 "Action is not allowed in the current phase.",
                 reward,
@@ -143,7 +169,12 @@ class CybersecurityOwaspEnvironment(
                 visible_test_result=visible_tests,
             )
         except Exception as exc:  # keep malformed agent actions from crashing the server
-            verifier, reward = evaluate_action(self._state, action, anti_cheat_flags)
+            verifier, reward = evaluate_action(
+                self._state,
+                action,
+                anti_cheat_flags,
+                invalid_action=True,
+            )
             return self._finish_step(
                 "Tool execution failed.",
                 reward,
@@ -164,91 +195,48 @@ class CybersecurityOwaspEnvironment(
     def _execute(
         self, action: CyberSecurityOWASPAction, anti_cheat_flags: list[str]
     ) -> tuple[str, dict, dict[str, float], str | None]:
-        verifier: dict = {"anti_cheat_flags": anti_cheat_flags}
-        reward = {key: 0.0 for key in (
-            "discovery",
-            "security",
-            "regression",
-            "public_routes",
-            "patch_quality",
-            "visible_tests",
-            "safety",
-            "anti_cheat",
-            "total",
-        )}
-        visible_tests = None
-        args = action.arguments or {}
+        verifier, reward = evaluate_action(self._state, action, anti_cheat_flags)
 
-        if action.tool_name == "noop":
-            return "No operation.", verifier, reward, None
-        if action.tool_name == "inspect_policy_graph":
-            return json.dumps(self._visible_policy_hint, indent=2, sort_keys=True), verifier, reward, None
-        if action.tool_name == "list_routes":
-            return json.dumps(self._workspace_summary["routes"], indent=2), verifier, reward, None
-        if action.tool_name == "read_openapi":
-            return json.dumps(
-                {
-                    "openapi": "3.1.0",
-                    "info": {"title": "Generated invoices app", "version": "0.1.0"},
-                    "paths": {
-                        "/health": {"get": {"x-public": True}},
-                        "/invoices/{invoice_id}": {"get": {"x-public": False}},
-                    },
-                },
-                indent=2,
-            ), verifier, reward, None
-        if action.tool_name == "read_file":
-            path = self._resolve_path(str(args.get("path", "")))
-            return path.read_text(encoding="utf-8"), verifier, reward, None
-        if action.tool_name == "search_code":
-            return self._search_code(str(args.get("query", ""))), verifier, reward, None
-        if action.tool_name == "send_local_request":
-            if not is_local_route(str(args.get("path", ""))):
-                raise ValueError("send_local_request only accepts local route paths")
-            response = simulate_request(
+        if action.tool_name in {
+            "noop",
+            "inspect_policy_graph",
+            "list_routes",
+            "read_openapi",
+            "read_file",
+            "search_code",
+            "send_local_request",
+            "compare_identities",
+            "patch_file",
+        }:
+            result = ActionTools(
                 self._state,
-                str(args.get("method", "GET")),
-                str(args.get("path", "")),
-                args.get("user_id"),
-            )
-            return json.dumps(response, indent=2, sort_keys=True), verifier, reward, None
-        if action.tool_name == "compare_identities":
-            path = str(args.get("path", ""))
-            first = str(args.get("first_user_id", ""))
-            second = str(args.get("second_user_id", ""))
-            if not is_local_route(path):
-                raise ValueError("compare_identities only accepts local route paths")
-            response = {
-                "first": simulate_request(self._state, str(args.get("method", "GET")), path, first),
-                "second": simulate_request(self._state, str(args.get("method", "GET")), path, second),
-            }
-            return json.dumps(response, indent=2, sort_keys=True), verifier, reward, None
+                self._visible_policy_hint,
+                self._workspace_summary,
+            ).execute(action)
+            return result.message, verifier, reward, result.visible_test_result
         if action.tool_name == "submit_finding":
             verifier, reward = evaluate_action(self._state, action, anti_cheat_flags)
+            self._state.verification_summary = verifier
             if verifier.get("finding", {}).get("valid"):
                 self._state.finding_submitted = True
                 self._state.phase = "patch"
                 return "Finding accepted. Patch phase unlocked.", verifier, reward, None
             return "Finding was not specific enough to unlock patching.", verifier, reward, None
-        if action.tool_name == "patch_file":
-            path = self._resolve_path(str(args.get("path", "")), write=True)
-            if "content" in args:
-                path.write_text(str(args["content"]), encoding="utf-8")
-            else:
-                self._apply_unified_diff(path, str(args.get("diff", "")))
-            return f"Patched {args.get('path')}.", verifier, reward, None
         if action.tool_name == "run_visible_tests":
             verifier, reward = evaluate_action(self._state, action, anti_cheat_flags)
+            self._state.verification_summary = verifier
             visible_tests = json.dumps(verifier.get("visible", {}), indent=2, sort_keys=True)
             return visible_tests, verifier, reward, visible_tests
         if action.tool_name == "submit_fix":
             verifier, reward = evaluate_action(self._state, action, anti_cheat_flags)
+            self._state.verification_summary = verifier
             self._state.patch_submitted = True
             security = verifier.get("security", {}).get("passed", False)
+            oracle = verifier.get("oracle_matrix", {}).get("passed", False)
             regression = verifier.get("regression", {}).get("passed", False)
             public = verifier.get("public_routes", {}).get("passed", False)
             quality = verifier.get("patch_quality", {}).get("passed", False)
-            self._state.success = bool(security and regression and public and quality)
+            self._state.success = bool(security and oracle and regression and public and quality)
             self._state.done = True
             self._state.phase = "done"
             self._state.failure_reason = None if self._state.success else "hidden_verifier_failed"
@@ -281,7 +269,10 @@ class CybersecurityOwaspEnvironment(
             visible_test_result=visible_test_result,
             done_reason=self._state.failure_reason,
         )
+        observation_record = obs.model_dump()
+        self._state.observation_history.append(observation_record)
         if self._state.done:
+            self._finalize_terminal_episode(observation_record)
             self._last_done_observation = obs
         return obs
 
@@ -314,53 +305,15 @@ class CybersecurityOwaspEnvironment(
             metadata={"episode_id": self._state.episode_id, "step_count": self._state.step_count},
         )
 
-    def _resolve_path(self, path: str, *, write: bool = False) -> Path:
-        allowed, normalized_or_error = is_path_allowed(self._state, path, write=write)
-        if not allowed:
-            raise ValueError(normalized_or_error)
-        return Path(str(self._state.hidden_facts["workspace"])) / normalized_or_error
-
-    def _search_code(self, query: str) -> str:
-        if not query:
-            raise ValueError("query is required")
-        results: list[str] = []
-        workspace = Path(str(self._state.hidden_facts["workspace"]))
-        for rel in self._state.hidden_facts.get("editable_files", []):
-            path = workspace / rel
-            text = path.read_text(encoding="utf-8")
-            for idx, line in enumerate(text.splitlines(), start=1):
-                if query.lower() in line.lower():
-                    results.append(f"{rel}:{idx}: {line}")
-        return "\n".join(results) or "No matches."
-
-    def _apply_unified_diff(self, path: Path, diff: str) -> None:
-        if not diff.strip():
-            raise ValueError("diff or content is required")
-        original = path.read_text(encoding="utf-8").splitlines(True)
-        output: list[str] = []
-        old_index = 0
-        lines = diff.splitlines(True)
-        i = 0
-        while i < len(lines):
-            line = lines[i]
-            if not line.startswith("@@"):
-                i += 1
-                continue
-            old_start = int(line.split()[1].split(",")[0][1:])
-            output.extend(original[old_index : old_start - 1])
-            old_index = old_start - 1
-            i += 1
-            while i < len(lines) and not lines[i].startswith("@@"):
-                hunk_line = lines[i]
-                if hunk_line.startswith(" "):
-                    output.append(original[old_index])
-                    old_index += 1
-                elif hunk_line.startswith("-"):
-                    old_index += 1
-                elif hunk_line.startswith("+"):
-                    output.append(hunk_line[1:])
-                elif hunk_line.startswith("\\"):
-                    pass
-                i += 1
-        output.extend(original[old_index:])
-        path.write_text("".join(output), encoding="utf-8")
+    def _finalize_terminal_episode(self, observation_record: dict[str, Any]) -> None:
+        if self._state.episode_artifact_path:
+            return
+        mastery = self._curriculum.record_episode(self._state)
+        self._state.curriculum_snapshot = {
+            **self._state.curriculum_snapshot,
+            "post_episode_mastery": mastery,
+        }
+        self._episode_logger.log_episode(
+            self._state,
+            final_observation=observation_record,
+        )

server/__init__.py

@@ -6,6 +6,16 @@
 
 """Cybersecurity Owasp environment server components."""
 
+from .adversarial_designer import BoundedAdversarialDesigner
 from .CyberSecurity_OWASP_environment import CybersecurityOwaspEnvironment
+from .curriculum import CurriculumController
+from .scenario_factory import ScenarioFactory
+from .verifier import MultiLayerVerifier
 
-__all__ = ["CybersecurityOwaspEnvironment"]
+__all__ = [
+    "BoundedAdversarialDesigner",
+    "CurriculumController",
+    "CybersecurityOwaspEnvironment",
+    "MultiLayerVerifier",
+    "ScenarioFactory",
+]

server/action_tools.py

@@ -0,0 +1,73 @@
+"""Typed action tool dispatcher for the generated app sandbox."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+
+try:
+    from ..models import CyberSecurityOWASPAction, CyberSecurityOWASPState
+    from .app_sandbox import AppSandbox
+except ImportError:  # pragma: no cover
+    from models import CyberSecurityOWASPAction, CyberSecurityOWASPState
+    from server.app_sandbox import AppSandbox
+
+
+@dataclass(frozen=True)
+class ToolResult:
+    message: str
+    visible_test_result: str | None = None
+
+
+class ActionTools:
+    """Executes phase-gated, safe tools against one episode state."""
+
+    def __init__(
+        self,
+        state: CyberSecurityOWASPState,
+        visible_policy_hint: dict,
+        workspace_summary: dict,
+    ):
+        self.state = state
+        self.visible_policy_hint = visible_policy_hint
+        self.workspace_summary = workspace_summary
+        self.sandbox = AppSandbox(state)
+
+    def execute(self, action: CyberSecurityOWASPAction) -> ToolResult:
+        args = action.arguments or {}
+        if action.tool_name == "noop":
+            return ToolResult("No operation.")
+        if action.tool_name == "inspect_policy_graph":
+            return ToolResult(json.dumps(self.visible_policy_hint, indent=2, sort_keys=True))
+        if action.tool_name == "list_routes":
+            return ToolResult(json.dumps(self.workspace_summary["routes"], indent=2))
+        if action.tool_name == "read_openapi":
+            return ToolResult(self.sandbox.read_openapi())
+        if action.tool_name == "read_file":
+            return ToolResult(self.sandbox.read_file(str(args.get("path", ""))))
+        if action.tool_name == "search_code":
+            return ToolResult(self.sandbox.search_code(str(args.get("query", ""))))
+        if action.tool_name == "send_local_request":
+            response = self.sandbox.send_local_request(
+                str(args.get("method", "GET")),
+                str(args.get("path", "")),
+                args.get("user_id"),
+            )
+            return ToolResult(json.dumps(response, indent=2, sort_keys=True))
+        if action.tool_name == "compare_identities":
+            response = self.sandbox.compare_identities(
+                str(args.get("method", "GET")),
+                str(args.get("path", "")),
+                str(args.get("first_user_id", "")),
+                str(args.get("second_user_id", "")),
+            )
+            return ToolResult(json.dumps(response, indent=2, sort_keys=True))
+        if action.tool_name == "patch_file":
+            result = self.sandbox.patch_file(
+                str(args.get("path", "")),
+                content=str(args["content"]) if "content" in args else None,
+                diff=str(args.get("diff", "")) if "content" not in args else None,
+            )
+            changed = "no diff" if not result["diff"].strip() else "diff recorded"
+            return ToolResult(f"Patched {result['path']} ({changed}).")
+        raise ValueError(f"Unhandled tool {action.tool_name}")

server/adversarial_designer.py

@@ -0,0 +1,59 @@
+"""Bounded adversarial scenario targeting for synthetic local lab episodes."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+try:
+    from .curriculum import WEAKNESS_TARGETS
+except ImportError:  # pragma: no cover
+    from server.curriculum import WEAKNESS_TARGETS
+
+
+TARGET_SPECS: dict[str, dict[str, Any]] = {
+    "same_role_cross_object": {
+        "description": "Same-role actor tries to read another user's object.",
+        "hidden_focus": ["same_tenant_other_user_blocked"],
+    },
+    "cross_tenant_boundary": {
+        "description": "Tenant-local admin is denied access to another tenant's resource.",
+        "hidden_focus": ["cross_tenant_blocked"],
+    },
+    "public_route_overlock": {
+        "description": "Public health route must remain unauthenticated after patching.",
+        "hidden_focus": ["health_public"],
+    },
+    "alternate_route_same_service": {
+        "description": "Alternate route/service access should follow the same policy oracle.",
+        "hidden_focus": ["oracle_matrix"],
+    },
+    "visible_test_edge_case": {
+        "description": "Visible tests are insufficient; hidden policy matrix decides success.",
+        "hidden_focus": ["visible_test_only_guard"],
+    },
+}
+
+
+@dataclass(frozen=True)
+class BoundedAdversarialDesigner:
+    """Chooses safe local lab variants that target tracked agent weaknesses."""
+
+    def design(self, *, seed: int, split: str, curriculum_profile: dict[str, Any]) -> dict[str, Any]:
+        target = str(curriculum_profile.get("target_weakness") or "")
+        if target not in TARGET_SPECS:
+            target = WEAKNESS_TARGETS[int(seed) % len(WEAKNESS_TARGETS)]
+        family = f"invoices.bola_idor.{target}"
+        if split == "hidden_eval":
+            family = f"heldout.{family}"
+        spec = TARGET_SPECS[target]
+        return {
+            "domain": "invoices",
+            "bug_family": "bola_idor",
+            "template_id": "fastapi_basic",
+            "scenario_family": family,
+            "target_weakness": target,
+            "hidden_focus": list(spec["hidden_focus"]),
+            "description": spec["description"],
+            "safe_lab_only": True,
+        }

server/app_sandbox.py

@@ -0,0 +1,139 @@
+"""Ephemeral generated app sandbox operations."""
+
+from __future__ import annotations
+
+import difflib
+import json
+from pathlib import Path
+from typing import Any
+
+try:
+    from ..models import CyberSecurityOWASPState
+    from ..safety import is_local_route
+    from ..validators import is_path_allowed, simulate_request
+except ImportError:  # pragma: no cover
+    from models import CyberSecurityOWASPState
+    from safety import is_local_route
+    from validators import is_path_allowed, simulate_request
+
+
+class AppSandbox:
+    """Encapsulates all generated workspace reads, patches, and local requests."""
+
+    def __init__(self, state: CyberSecurityOWASPState):
+        self.state = state
+
+    @property
+    def workspace(self) -> Path:
+        return Path(str(self.state.hidden_facts["workspace"]))
+
+    def read_file(self, path: str) -> str:
+        return self._resolve_path(path).read_text(encoding="utf-8")
+
+    def search_code(self, query: str) -> str:
+        if not query:
+            raise ValueError("query is required")
+        results: list[str] = []
+        for rel in self.state.hidden_facts.get("editable_files", []):
+            path = self.workspace / rel
+            text = path.read_text(encoding="utf-8")
+            for idx, line in enumerate(text.splitlines(), start=1):
+                if query.lower() in line.lower():
+                    results.append(f"{rel}:{idx}: {line}")
+        return "\n".join(results) or "No matches."
+
+    def patch_file(self, path: str, *, content: str | None = None, diff: str | None = None) -> dict[str, str]:
+        target = self._resolve_path(path, write=True)
+        before = target.read_text(encoding="utf-8")
+        if content is not None:
+            target.write_text(content, encoding="utf-8")
+        else:
+            self._apply_unified_diff(target, diff or "")
+        after = target.read_text(encoding="utf-8")
+        patch_diff = "".join(
+            difflib.unified_diff(
+                before.splitlines(True),
+                after.splitlines(True),
+                fromfile=path,
+                tofile=path,
+            )
+        )
+        self.state.patch_diff = patch_diff
+        files_touched = self.state.metrics.setdefault("files_touched", [])
+        if path not in files_touched:
+            files_touched.append(path)
+        return {"path": path, "diff": patch_diff}
+
+    def read_openapi(self) -> str:
+        routes = self.state.visible_facts.get("workspace_summary", {}).get("routes", [])
+        paths: dict[str, Any] = {}
+        for route in routes:
+            paths.setdefault(route["path"], {})[route["method"].lower()] = {
+                "x-public": bool(route.get("public", False))
+            }
+        return json.dumps(
+            {
+                "openapi": "3.1.0",
+                "info": {"title": "Generated invoices app", "version": "0.1.0"},
+                "paths": paths,
+            },
+            indent=2,
+            sort_keys=True,
+        )
+
+    def send_local_request(self, method: str, path: str, user_id: str | None = None) -> dict[str, Any]:
+        if not is_local_route(path):
+            raise ValueError("send_local_request only accepts local route paths")
+        return simulate_request(self.state, method, path, user_id)
+
+    def compare_identities(
+        self,
+        method: str,
+        path: str,
+        first_user_id: str,
+        second_user_id: str,
+    ) -> dict[str, Any]:
+        if not is_local_route(path):
+            raise ValueError("compare_identities only accepts local route paths")
+        return {
+            "first": simulate_request(self.state, method, path, first_user_id),
+            "second": simulate_request(self.state, method, path, second_user_id),
+        }
+
+    def _resolve_path(self, path: str, *, write: bool = False) -> Path:
+        allowed, normalized_or_error = is_path_allowed(self.state, path, write=write)
+        if not allowed:
+            raise ValueError(normalized_or_error)
+        return self.workspace / normalized_or_error
+
+    def _apply_unified_diff(self, path: Path, diff: str) -> None:
+        if not diff.strip():
+            raise ValueError("diff or content is required")
+        original = path.read_text(encoding="utf-8").splitlines(True)
+        output: list[str] = []
+        old_index = 0
+        lines = diff.splitlines(True)
+        i = 0
+        while i < len(lines):
+            line = lines[i]
+            if not line.startswith("@@"):
+                i += 1
+                continue
+            old_start = int(line.split()[1].split(",")[0][1:])
+            output.extend(original[old_index : old_start - 1])
+            old_index = old_start - 1
+            i += 1
+            while i < len(lines) and not lines[i].startswith("@@"):
+                hunk_line = lines[i]
+                if hunk_line.startswith(" "):
+                    output.append(original[old_index])
+                    old_index += 1
+                elif hunk_line.startswith("-"):
+                    old_index += 1
+                elif hunk_line.startswith("+"):
+                    output.append(hunk_line[1:])
+                elif hunk_line.startswith("\\"):
+                    pass
+                i += 1
+        output.extend(original[old_index:])
+        path.write_text("".join(output), encoding="utf-8")

server/authz_oracle.py

@@ -0,0 +1,92 @@
+"""Policy-oracle matrix for deterministic authorization verification."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+try:
+    from ..models import CyberSecurityOWASPState
+    from ..validators import simulate_request
+except ImportError:  # pragma: no cover
+    from models import CyberSecurityOWASPState
+    from validators import simulate_request
+
+
+@dataclass(frozen=True)
+class AuthzOracle:
+    """Builds hidden allowed/denied tuples from policy facts."""
+
+    def matrix(self, state: CyberSecurityOWASPState) -> list[dict[str, Any]]:
+        hidden = state.hidden_facts
+        return [
+            {
+                "name": "owner_can_read_owned_invoice",
+                "method": "GET",
+                "path": f"/invoices/{hidden['owner_invoice_id']}",
+                "user_id": hidden["owner_user_id"],
+                "expected_status": 200,
+                "allowed": True,
+            },
+            {
+                "name": "resource_owner_can_read_own_invoice",
+                "method": "GET",
+                "path": f"/invoices/{hidden['other_invoice_id']}",
+                "user_id": hidden["intruder_user_id"],
+                "expected_status": 200,
+                "allowed": True,
+            },
+            {
+                "name": "billing_admin_can_read_same_tenant_invoice",
+                "method": "GET",
+                "path": f"/invoices/{hidden['other_invoice_id']}",
+                "user_id": hidden["admin_user_id"],
+                "expected_status": 200,
+                "allowed": True,
+            },
+            {
+                "name": "same_tenant_non_owner_denied",
+                "method": "GET",
+                "path": f"/invoices/{hidden['other_invoice_id']}",
+                "user_id": hidden["owner_user_id"],
+                "expected_status": 403,
+                "allowed": False,
+            },
+            {
+                "name": "cross_tenant_admin_denied",
+                "method": "GET",
+                "path": f"/invoices/{hidden['foreign_invoice_id']}",
+                "user_id": hidden["admin_user_id"],
+                "expected_status": 403,
+                "allowed": False,
+            },
+            {
+                "name": "health_remains_public",
+                "method": "GET",
+                "path": "/health",
+                "user_id": None,
+                "expected_status": 200,
+                "allowed": True,
+            },
+        ]
+
+    def evaluate(self, state: CyberSecurityOWASPState) -> dict[str, Any]:
+        cases = []
+        for case in self.matrix(state):
+            response = simulate_request(
+                state,
+                str(case["method"]),
+                str(case["path"]),
+                case.get("user_id"),
+            )
+            actual = int(response["status"])
+            cases.append(
+                {
+                    "name": case["name"],
+                    "allowed": bool(case["allowed"]),
+                    "expected_status": int(case["expected_status"]),
+                    "actual_status": actual,
+                    "passed": actual == int(case["expected_status"]),
+                }
+            )
+        return {"passed": all(case["passed"] for case in cases), "cases": cases}

server/curriculum.py

@@ -0,0 +1,99 @@
+"""Runtime curriculum controller for closed-loop scenario selection."""
+
+from __future__ import annotations
+
+from collections import defaultdict, deque
+from dataclasses import dataclass, field
+from typing import Any
+
+try:
+    from ..models import CyberSecurityOWASPState
+except ImportError:  # pragma: no cover
+    from models import CyberSecurityOWASPState
+
+
+DIFFICULTY_TIERS = ("warmup", "beginner", "intermediate", "advanced", "expert")
+WEAKNESS_TARGETS = (
+    "same_role_cross_object",
+    "cross_tenant_boundary",
+    "public_route_overlock",
+    "alternate_route_same_service",
+    "visible_test_edge_case",
+)
+
+
+@dataclass
+class CurriculumController:
+    """Tracks episode outcomes and picks the next bounded weakness target."""
+
+    window_size: int = 10
+    reward_trend: deque[float] = field(default_factory=lambda: deque(maxlen=10))
+    outcomes_by_target: dict[str, list[bool]] = field(default_factory=lambda: defaultdict(list))
+    failures_by_target: dict[str, int] = field(default_factory=lambda: defaultdict(int))
+    episodes_seen: int = 0
+
+    def select_profile(
+        self,
+        *,
+        seed: int,
+        split: str = "train",
+        requested_difficulty: int = 0,
+    ) -> dict[str, Any]:
+        difficulty = self._difficulty_for_split(split, requested_difficulty)
+        target = self._target_for_seed(seed, split)
+        if self.failures_by_target:
+            target = max(
+                WEAKNESS_TARGETS,
+                key=lambda item: (self.failures_by_target.get(item, 0), -WEAKNESS_TARGETS.index(item)),
+            )
+        return {
+            "difficulty": difficulty,
+            "difficulty_tier": DIFFICULTY_TIERS[min(difficulty, len(DIFFICULTY_TIERS) - 1)],
+            "target_weakness": target,
+            "split": split,
+            "episodes_seen": self.episodes_seen,
+            "recent_reward_mean": self._recent_reward_mean(),
+            "mastery": self.mastery_snapshot(),
+        }
+
+    def record_episode(self, state: CyberSecurityOWASPState) -> dict[str, Any]:
+        target = state.target_weakness or "same_role_cross_object"
+        success = bool(state.success)
+        self.episodes_seen += 1
+        self.outcomes_by_target[target].append(success)
+        if not success:
+            self.failures_by_target[target] += 1
+        self.reward_trend.append(float(state.last_reward or 0.0))
+        return self.mastery_snapshot()
+
+    def mastery_snapshot(self) -> dict[str, Any]:
+        target_mastery = {}
+        for target in WEAKNESS_TARGETS:
+            outcomes = self.outcomes_by_target.get(target, [])
+            target_mastery[target] = {
+                "episodes": len(outcomes),
+                "success_rate": sum(1 for item in outcomes if item) / max(1, len(outcomes)),
+                "failures": self.failures_by_target.get(target, 0),
+            }
+        return {
+            "episodes_seen": self.episodes_seen,
+            "recent_reward_mean": self._recent_reward_mean(),
+            "target_mastery": target_mastery,
+        }
+
+    def _difficulty_for_split(self, split: str, requested_difficulty: int) -> int:
+        difficulty = max(0, min(int(requested_difficulty), len(DIFFICULTY_TIERS) - 1))
+        if split == "hidden_eval":
+            return max(3, difficulty)
+        if self.episodes_seen >= self.window_size and self._recent_reward_mean() > 10.0:
+            return min(difficulty + 1, len(DIFFICULTY_TIERS) - 1)
+        return difficulty
+
+    def _target_for_seed(self, seed: int, split: str) -> str:
+        offset = 2 if split == "hidden_eval" else 0
+        return WEAKNESS_TARGETS[(int(seed) + offset) % len(WEAKNESS_TARGETS)]
+
+    def _recent_reward_mean(self) -> float:
+        if not self.reward_trend:
+            return 0.0
+        return sum(self.reward_trend) / len(self.reward_trend)

server/episode_logger.py

@@ -0,0 +1,66 @@
+"""Episode artifact logging for training, debugging, and demos."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+try:
+    from ..models import CyberSecurityOWASPState
+except ImportError:  # pragma: no cover
+    from models import CyberSecurityOWASPState
+
+
+class EpisodeArtifactLogger:
+    """Appends compact JSONL episode transcripts under outputs/rollouts."""
+
+    def __init__(self, output_path: str | Path | None = None):
+        configured = output_path or os.getenv("CYBERSECURITY_OWASP_EPISODE_LOG")
+        self.output_path = Path(configured) if configured else Path("outputs/rollouts/episodes.jsonl")
+
+    def log_episode(
+        self,
+        state: CyberSecurityOWASPState,
+        *,
+        final_observation: dict[str, Any] | None = None,
+    ) -> Path:
+        self.output_path.parent.mkdir(parents=True, exist_ok=True)
+        record = {
+            "episode_id": state.episode_id,
+            "task_id": state.task_id,
+            "seed": state.seed,
+            "split": state.split,
+            "difficulty": state.difficulty,
+            "difficulty_tier": state.difficulty_tier,
+            "template_id": state.template_id,
+            "scenario_family": state.scenario_family,
+            "domain": state.domain,
+            "bug_family": state.bug_family,
+            "target_weakness": state.target_weakness,
+            "agent_actions": state.action_history,
+            "observations": state.observation_history,
+            "final_observation": final_observation or {},
+            "patch_diff": state.patch_diff,
+            "visible_test_result": self._verifier_layer(state, "visible"),
+            "hidden_test_result": self._verifier_layer(state, "hidden_tests"),
+            "oracle_result": self._verifier_layer(state, "oracle_matrix"),
+            "regression_result": self._verifier_layer(state, "regression"),
+            "reward_breakdown": state.reward_history[-1] if state.reward_history else {},
+            "reward_breakdown_by_step": state.reward_history,
+            "final_status": "resolved" if state.success else "failed",
+            "failure_reason": state.failure_reason,
+            "safety_violations": [
+                flag for flag in state.anti_cheat_flags if "network" in flag or "unsafe" in flag
+            ],
+            "anti_cheat_flags": state.anti_cheat_flags,
+            "metrics": state.metrics,
+        }
+        with self.output_path.open("a", encoding="utf-8") as handle:
+            handle.write(json.dumps(record, sort_keys=True) + "\n")
+        state.episode_artifact_path = str(self.output_path)
+        return self.output_path
+
+    def _verifier_layer(self, state: CyberSecurityOWASPState, key: str) -> Any:
+        return (state.verification_summary or {}).get(key)

server/reward_engine.py

@@ -5,45 +5,24 @@ from __future__ import annotations
 try:
     from ..models import CyberSecurityOWASPAction, CyberSecurityOWASPState
     from ..rewards import compute_reward
-    from ..validators import (
-        patch_quality,
-        run_hidden_regression_tests,
-        run_hidden_security_tests,
-        run_public_route_tests,
-        run_visible_tests,
-        verify_finding,
-    )
+    from .verifier import MultiLayerVerifier
 except ImportError:  # pragma: no cover
     from models import CyberSecurityOWASPAction, CyberSecurityOWASPState
     from rewards import compute_reward
-    from validators import (
-        patch_quality,
-        run_hidden_regression_tests,
-        run_hidden_security_tests,
-        run_public_route_tests,
-        run_visible_tests,
-        verify_finding,
-    )
+    from server.verifier import MultiLayerVerifier
 
 
 def evaluate_action(
     state: CyberSecurityOWASPState,
     action: CyberSecurityOWASPAction,
     anti_cheat_flags: list[str] | None = None,
+    *,
+    invalid_action: bool = False,
 ) -> tuple[dict, dict[str, float]]:
-    verifier_result: dict = {"anti_cheat_flags": anti_cheat_flags or []}
-    if action.tool_name == "submit_finding":
-        verifier_result["finding"] = verify_finding(state, action.arguments)
-    elif action.tool_name == "run_visible_tests":
-        verifier_result["visible"] = run_visible_tests(state)
-    elif action.tool_name == "submit_fix":
-        verifier_result.update(
-            {
-                "visible": run_visible_tests(state),
-                "security": run_hidden_security_tests(state),
-                "regression": run_hidden_regression_tests(state),
-                "public_routes": run_public_route_tests(state),
-                "patch_quality": patch_quality(state),
-            }
-        )
+    verifier_result = MultiLayerVerifier().evaluate_action(
+        state,
+        action,
+        anti_cheat_flags,
+        invalid_action=invalid_action,
+    )
     return verifier_result, compute_reward(state, action, verifier_result)

server/scenario_factory.py

@@ -0,0 +1,134 @@
+"""Closed-loop scenario factory for CyberSecurity_OWASP."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+try:
+    from ..fixture_generator import visible_workspace_summary
+    from ..policy_graph import build_invoice_policy
+    from ..template_renderer import render_fastapi_basic
+    from .adversarial_designer import BoundedAdversarialDesigner
+except ImportError:  # pragma: no cover
+    from fixture_generator import visible_workspace_summary
+    from policy_graph import build_invoice_policy
+    from template_renderer import render_fastapi_basic
+    from server.adversarial_designer import BoundedAdversarialDesigner
+
+
+def _make_workspace(prefix: str) -> Path:
+    root = Path(os.getenv("CYBERSECURITY_OWASP_WORKSPACE_ROOT", tempfile.gettempdir()))
+    root.mkdir(parents=True, exist_ok=True)
+    for _ in range(100):
+        workspace = root / f"{prefix}{uuid4().hex[:12]}"
+        try:
+            workspace.mkdir()
+        except FileExistsError:
+            continue
+        return workspace
+    raise RuntimeError("Unable to create isolated scenario workspace")
+
+
+def _visible_policy_hint(public_hint: dict[str, Any]) -> dict[str, Any]:
+    """Return partial policy observability without hidden oracle/test labels."""
+
+    return {
+        "domain": public_hint.get("domain", "invoices"),
+        "policy_rules": list(public_hint.get("policy_rules", [])),
+        "fixture_aliases": {
+            "users": dict(public_hint.get("users", {})),
+            "resources": dict(public_hint.get("resources", {})),
+        },
+        "public_routes": list(public_hint.get("public_routes", [])),
+        "observation_contract": {
+            "visible": [
+                "product policy summary",
+                "fixture aliases needed for local requests",
+                "route summaries",
+                "visible test results",
+            ],
+            "hidden": [
+                "evaluator-only policy tuples",
+                "withheld invariant checks",
+                "withheld scenario labels",
+                "held-out family label",
+            ],
+        },
+    }
+
+
+class ScenarioFactory:
+    """Compiles deterministic local app scenarios from curriculum profiles."""
+
+    def __init__(self, designer: BoundedAdversarialDesigner | None = None):
+        self.designer = designer or BoundedAdversarialDesigner()
+
+    def compile_scenario(
+        self,
+        seed: int,
+        *,
+        split: str = "train",
+        difficulty: int = 0,
+        curriculum_profile: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        profile = curriculum_profile or {
+            "difficulty": difficulty,
+            "difficulty_tier": "warmup",
+            "target_weakness": "same_role_cross_object",
+        }
+        adversarial_spec = self.designer.design(
+            seed=seed, split=split, curriculum_profile=profile
+        )
+        compiled = build_invoice_policy(seed)
+        workspace = _make_workspace(prefix=f"cybersecurity_owasp_{split}_{seed}_")
+        public_hint = _visible_policy_hint(compiled.public_hint)
+        editable_files = render_fastapi_basic(workspace, public_hint, compiled.hidden_facts)
+        workspace_summary = visible_workspace_summary(editable_files, public_hint)
+        workspace_summary.update(
+            {
+                "template_id": adversarial_spec["template_id"],
+                "target_weakness": adversarial_spec["target_weakness"],
+            }
+        )
+
+        hidden = dict(compiled.hidden_facts)
+        hidden.update(
+            {
+                "workspace": str(workspace),
+                "editable_files": editable_files,
+                "initial_file_hashes": {
+                    path: (workspace / path).read_text(encoding="utf-8")
+                    for path in editable_files
+                },
+                "adversarial_spec": adversarial_spec,
+                "scenario_family": adversarial_spec["scenario_family"],
+                "template_id": adversarial_spec["template_id"],
+                "target_weakness": adversarial_spec["target_weakness"],
+                "oracle_hidden_focus": adversarial_spec["hidden_focus"],
+            }
+        )
+
+        return {
+            "task_id": f"{split}-invoices-bola-{seed}",
+            "workspace": workspace,
+            "domain": adversarial_spec["domain"],
+            "bug_family": adversarial_spec["bug_family"],
+            "scenario_family": adversarial_spec["scenario_family"],
+            "template_id": adversarial_spec["template_id"],
+            "target_weakness": adversarial_spec["target_weakness"],
+            "difficulty": int(profile.get("difficulty", difficulty)),
+            "difficulty_tier": str(profile.get("difficulty_tier", "warmup")),
+            "curriculum_snapshot": profile,
+            "task_brief": (
+                "Inspect the generated invoices app and policy. Find the broken "
+                "authorization behavior, submit a finding with local evidence, patch "
+                "the app, preserve intended owner/admin/public behavior, then submit."
+            ),
+            "public_hint": public_hint,
+            "workspace_summary": workspace_summary,
+            "hidden_facts": hidden,
+        }

server/verifier.py

@@ -0,0 +1,81 @@
+"""Multi-layer deterministic verifier for CyberSecurity_OWASP."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any
+
+try:
+    from ..models import CyberSecurityOWASPAction, CyberSecurityOWASPState
+    from ..validators import (
+        patch_quality,
+        run_hidden_regression_tests,
+        run_hidden_security_tests,
+        run_public_route_tests,
+        run_visible_tests,
+        verify_finding,
+    )
+    from .authz_oracle import AuthzOracle
+except ImportError:  # pragma: no cover
+    from models import CyberSecurityOWASPAction, CyberSecurityOWASPState
+    from validators import (
+        patch_quality,
+        run_hidden_regression_tests,
+        run_hidden_security_tests,
+        run_public_route_tests,
+        run_visible_tests,
+        verify_finding,
+    )
+    from server.authz_oracle import AuthzOracle
+
+
+@dataclass
+class MultiLayerVerifier:
+    """Aggregates visible, hidden, oracle, regression, and patch-quality checks."""
+
+    oracle: AuthzOracle = AuthzOracle()
+
+    def evaluate_action(
+        self,
+        state: CyberSecurityOWASPState,
+        action: CyberSecurityOWASPAction,
+        anti_cheat_flags: list[str] | None = None,
+        *,
+        invalid_action: bool = False,
+    ) -> dict[str, Any]:
+        verifier_result: dict[str, Any] = {
+            "anti_cheat_flags": anti_cheat_flags or [],
+            "invalid_action": invalid_action,
+            "repeated_action": self._is_repeated_action(state, action),
+        }
+        if action.tool_name == "submit_finding":
+            verifier_result["finding"] = verify_finding(state, action.arguments)
+        elif action.tool_name == "run_visible_tests":
+            verifier_result["visible"] = run_visible_tests(state)
+        elif action.tool_name == "submit_fix":
+            verifier_result.update(self.run_terminal_checks(state))
+        return verifier_result
+
+    def run_terminal_checks(self, state: CyberSecurityOWASPState) -> dict[str, Any]:
+        security = run_hidden_security_tests(state)
+        return {
+            "visible": run_visible_tests(state),
+            "hidden_tests": security,
+            "security": security,
+            "oracle_matrix": self.oracle.evaluate(state),
+            "regression": run_hidden_regression_tests(state),
+            "public_routes": run_public_route_tests(state),
+            "patch_quality": patch_quality(state),
+        }
+
+    def public_summary(self, verifier_result: dict[str, Any]) -> dict[str, Any]:
+        """Return verifier fields that are safe for state/debug summaries."""
+
+        return json.loads(json.dumps(verifier_result))
+
+    def _is_repeated_action(
+        self, state: CyberSecurityOWASPState, action: CyberSecurityOWASPAction
+    ) -> bool:
+        current = {"tool_name": action.tool_name, "arguments": action.arguments}
+        return sum(1 for item in state.action_history if item == current) > 1

tests/test_closed_loop_runtime.py

@@ -0,0 +1,94 @@
+import json
+from pathlib import Path
+
+from CyberSecurity_OWASP.models import CyberSecurityOWASPAction
+from CyberSecurity_OWASP.server.adversarial_designer import BoundedAdversarialDesigner
+from CyberSecurity_OWASP.server.authz_oracle import AuthzOracle
+from CyberSecurity_OWASP.server.curriculum import CurriculumController
+from CyberSecurity_OWASP.server.verifier import MultiLayerVerifier
+
+from .helpers import apply_secure_patch, make_env, submit_valid_finding
+
+
+def test_curriculum_selects_profile_and_tracks_mastery():
+    controller = CurriculumController()
+    profile = controller.select_profile(seed=3, split="train", requested_difficulty=1)
+
+    assert profile["difficulty_tier"] == "beginner"
+    assert profile["target_weakness"]
+    assert "target_mastery" in profile["mastery"]
+
+    env = make_env(70)
+    controller.record_episode(env.state)
+    snapshot = controller.mastery_snapshot()
+    assert snapshot["episodes_seen"] == 1
+
+
+def test_adversarial_designer_marks_hidden_eval_as_heldout_family():
+    designer = BoundedAdversarialDesigner()
+    spec = designer.design(
+        seed=4,
+        split="hidden_eval",
+        curriculum_profile={"target_weakness": "cross_tenant_boundary"},
+    )
+
+    assert spec["safe_lab_only"] is True
+    assert spec["scenario_family"].startswith("heldout.")
+    assert spec["target_weakness"] == "cross_tenant_boundary"
+
+
+def test_reset_records_scenario_family_and_partial_observability():
+    env = make_env(71)
+    obs = env.reset(seed=71, split="hidden_eval", difficulty=1)
+    serialized_hint = json.dumps(obs.visible_policy_hint).lower()
+
+    assert env.state.scenario_family.startswith("heldout.")
+    assert env.state.difficulty_tier in {"advanced", "expert"}
+    assert "oracle_matrix" not in serialized_hint
+    assert "hidden_tests" not in serialized_hint
+    assert "injected bug" not in serialized_hint
+
+
+def test_authz_oracle_fails_vulnerable_app_and_passes_secure_patch():
+    env = make_env(72)
+    oracle = AuthzOracle()
+
+    vulnerable = oracle.evaluate(env.state)
+    assert vulnerable["passed"] is False
+
+    submit_valid_finding(env)
+    apply_secure_patch(env)
+    fixed = oracle.evaluate(env.state)
+    assert fixed["passed"] is True
+
+
+def test_multilayer_verifier_aggregates_terminal_layers():
+    env = make_env(73)
+    submit_valid_finding(env)
+    apply_secure_patch(env)
+
+    verifier = MultiLayerVerifier().run_terminal_checks(env.state)
+    assert verifier["visible"]["passed"] is True
+    assert verifier["hidden_tests"]["passed"] is True
+    assert verifier["oracle_matrix"]["passed"] is True
+    assert verifier["regression"]["passed"] is True
+    assert verifier["public_routes"]["passed"] is True
+    assert verifier["patch_quality"]["passed"] is True
+
+
+def test_solved_episode_writes_jsonl_artifact_with_verifier_fields():
+    env = make_env(74)
+    submit_valid_finding(env)
+    apply_secure_patch(env)
+    env.step(CyberSecurityOWASPAction(tool_name="run_visible_tests"))
+    final = env.step(CyberSecurityOWASPAction(tool_name="submit_fix"))
+
+    artifact_path = Path(env.state.episode_artifact_path or "")
+    assert final.done is True
+    assert artifact_path.exists()
+    record = json.loads(artifact_path.read_text(encoding="utf-8").splitlines()[-1])
+    assert record["episode_id"] == env.state.episode_id
+    assert record["final_status"] == "resolved"
+    assert record["hidden_test_result"]["passed"] is True
+    assert record["oracle_result"]["passed"] is True
+    assert record["reward_breakdown"]["total"] >= 12.0

tests/test_web_interface.py

@@ -1,6 +1,6 @@
 from fastapi.testclient import TestClient
 
-from server.app import app
+from CyberSecurity_OWASP.server.app import app
 
 
 def test_space_root_redirects_to_openenv_web_ui():

training/rollout.py

@@ -67,6 +67,11 @@ def rollout_once(trainer, env, tokenizer=None, dataset_prompt: str = "", max_ste
 
     final_breakdown = getattr(observation, "reward_breakdown", {}) or {}
     state = env.state if not callable(getattr(env, "state", None)) else env.state()
+    verifier = getattr(state, "verification_summary", {}) or {}
+    anti_cheat_flags = getattr(state, "anti_cheat_flags", []) or []
+    invalid_actions = [
+        obs for obs in observation_trace if obs.get("last_action_valid") is False
+    ]
     return {
         "prompt_ids": prompt_ids,
         "completion_ids": completion_ids,
@@ -79,6 +84,16 @@ def rollout_once(trainer, env, tokenizer=None, dataset_prompt: str = "", max_ste
         "reward_anti_cheat": float(final_breakdown.get("anti_cheat", 0.0)),
         "success": bool(getattr(state, "success", False)),
         "episode_length": len(action_trace),
+        "exploit_blocked": bool((verifier.get("security") or {}).get("passed", False)),
+        "regression_preserved": bool((verifier.get("regression") or {}).get("passed", False)),
+        "public_routes_preserved": bool((verifier.get("public_routes") or {}).get("passed", False)),
+        "anti_cheat_pass": not bool(anti_cheat_flags),
+        "invalid_action_rate": len(invalid_actions) / max(1, len(action_trace)),
+        "timeout": getattr(state, "failure_reason", None) == "max_steps_exceeded",
+        "safety_violation": bool(
+            any("network" in flag or "unsafe" in flag for flag in anti_cheat_flags)
+        ),
+        "episode_artifact_path": getattr(state, "episode_artifact_path", None),
         "actions": action_trace,
         "observations": observation_trace,
     }