Add Rust-backed eval stack: scenario packs, adapter, spine, integration tests

- EVAL_STACK_PLAN.md: architecture/phasing; Rust-only backend, Bench
refactored in place to reuse OpenRA-RL-Training components.
- openra_bench/rust_adapter.py: Rust obs -> render/score schema with
delta-derived signals (Rust reward is hardcoded 0.0).
- openra_bench/eval_core.py: episode spine (run_episode/run_level) with
per-turn declarative win/fail evaluation.
- openra_bench/scenarios/: contributor scenario-pack layer (schema,
composable win-condition grammar, loader w/ Rust map gating, validate
CLI, CONTRIBUTING + TEMPLATE) + 6 authored P/R/A packs (3 levels each).
- tests/test_rust_integration.py: 17 rule-based tests booting the real
Rust engine (tool correctness, corner cases, determinism, win/fail
wiring, end-to-end pack runs).

EVAL_STACK_PLAN.md

@@ -0,0 +1,179 @@
+# OpenRA-Bench → Unified Eval Stack Plan
+
+## Goal
+
+Turn **OpenRA-Bench** into an efficient, customizable harness to evaluate model
+performance on **spatial, multi-modal, and complex multi-target multi-step
+reasoning / planning**, by reusing the mature eval stack from
+**OpenRA-RL-Training** on top of the **Rust** environment (OpenRA-Rust).
+
+Two evaluation modes:
+
+1. **Fixed-scenario eval** — N episodes of one model on a controlled scenario,
+   scored with verifiable rubrics + composite score + 95% CIs.
+2. **Pairwise adversarial 1v1** — two models in one 1v1 map, win-rate / Elo.
+
+Scenarios must be *controlled* and isolate specific capabilities (e.g. the
+observed failure: model cannot connect an unexplored region to a target area to
+explore — a vision/perception + spatial-planning deficiency).
+
+## Core framing: the Perception → Reasoning → Action chain
+
+Every scenario evaluates one chain: **read the (visual + symbolic) state →
+form a multi-step plan → emit valid commands that execute it**. The eval must
+*attribute* failure to a specific link, not just report a score:
+
+- **Perception** — did the model correctly read the minimap/state? (e.g.
+  locate the unexplored region and the target). Probe via state-readback /
+  forced-choice checks derived from ground-truth obs.
+- **Reasoning** — given correct perception, did it form a valid plan? (e.g.
+  connect unexplored→target, sequence multi-target order). Probe via plan
+  quality vs optimal (path length, target ordering, sub-goal coverage).
+- **Action** — did it emit syntactically/semantically valid commands that
+  realize the plan? Probe via action-validity rate and plan↔execution drift.
+
+Per-scenario rubrics carry one diagnostic per link so a low score points at the
+broken link. This is the primary product differentiator vs a raw win-rate bench.
+
+## Decisions (locked)
+
+- **Repo**: refactor OpenRA-Bench in place; it imports Training's eval stack.
+  OpenRA-RL-Training stays source of truth for the engine code.
+- **Backend**: Rust only (`openra_train` PyO3). C# (openra-rl) is slow/fragile
+  and is dropped from the eval path. Rust must be made faithful to the C#
+  reference where scenarios require it.
+- **Multi-modal**: reuse Training's `minimap_renderer.render_minimap()` PNG,
+  injected as `image_url` in the agent prompt.
+
+## Source components reused from OpenRA-RL-Training
+
+| Component | Path | Role in Bench |
+|---|---|---|
+| Episode engine | `openra_rl_training/training/agent_rollout.py` (`play_episodes_async`) | Drives the real model loop (currently Bench's agent fn is a no-op) |
+| Reward dims | `training/reward_funcs.py` | Per-scenario weighted scoring |
+| Rust pool | `training/rust_env_pool.py` | The only backend |
+| Minimap | `training/minimap_renderer.py` | Multi-modal observation |
+| Scenarios/rubrics | `scenarios/*.yaml`, `curricula/*.yaml`, verifiable metrics | Controlled tasks + pass/fail |
+| CI comparison | `scripts/build_eval_comparison.py` | Stat-sound model comparison |
+
+## Rust faithfulness gap (drives sequencing)
+
+- Commands: 3/22 (Move, Attack, Observe). Missing: Build/Train/Harvest/Deploy/
+  Sell/Repair/Stance/Transport/Power/RallyPoint/Guard/Patrol…
+- Observations: ~30% of C# proto. Missing: economy, production, military stats,
+  spatial tensor, kill_events, result/reward fields.
+- Scenarios: 2 hand-built (rush-hour, scout-maginot). No generic `.oramap` load.
+- Engine: movement (A*), combat, projectiles, fog, static defenses = done.
+  Economy, production/tech, transport, multi-armament = not done.
+
+→ Movement/combat/fog scenarios + combat-only 1v1 work **today**. Economy /
+production / tech scenarios require Rust engine work first.
+
+## CRITICAL FINDING (verified end-to-end, local)
+
+`play_episodes_async` (agent_rollout.py:4815) is **hardwired to the C# gRPC
+server** via `openra_env.mcp_ws_client.OpenRAMCPClient` and is entangled with
+TRL (tokenizer, prompt_ids/completion_ids, worker pool, partial cache). It is
+**not reusable as-is on Rust**. `rust_env_pool` is used only by the lighter
+`rollout.py` path.
+
+Also: `minimap_renderer.render_minimap()` expects `state["minimap"]` (ASCII),
+`units_summary`, `enemy_summary` — **none of which the Rust env emits.**
+
+Verified live Rust obs schema (`openra_train` rush-hour, local wheel,
+Python 3.12, anaconda):
+```
+keys = enemy_buildings_summary, enemy_hp, enemy_positions, explored_cells,
+       explored_percent, game_tick, unit_hp, unit_positions, units_killed
+unit_positions = {actor_id: {cell_x, cell_y[, target, activity, ...]}}
+step() -> (obs, reward=0.0 hardcoded, done:bool, info={game_tick, warnings})
+```
+No `minimap` ASCII, no economy/military/result/reward, no terrain.
+
+**Consequence:** Phase 0 builds a Bench-side episode loop that reuses
+*components* (reward_funcs, minimap_renderer, scenario loader, action parser)
+behind a **Rust→schema adapter** (`openra_bench/rust_adapter.py`). The adapter
+is the crux and overlaps directly with the "make Rust faithful" workstream:
+
+- `unit_positions` → `units_summary` (renderer/prompt schema)
+- `enemy_positions` + `enemy_buildings_summary` → `enemy_summary`
+- synthesize ASCII `minimap` from `explored_cells` + scenario map dims
+- load `terrain_png` from the scenario's base `.oramap` (as Training does)
+- derive scoring signals (kills, discovery, exploration, outcome) from obs
+  deltas since Rust `reward` is hardcoded 0.0 — feeds reward_funcs + the
+  P/R/A diagnostics directly.
+
+## Target Bench layout
+
+```
+OpenRA-Bench/
+  openra_bench/
+    eval_core.py      # thin wrapper over play_episodes_async, Rust backend forced
+    agent.py          # REAL model agent (OpenAI-compatible), minimap multimodal
+    scenarios/        # controlled eval scenarios (symlink/copy + Bench-authored)
+    rubrics.py        # verifiable + composite scoring (reuse Training)
+    pairwise.py       # 1v1 adversarial orchestration + Elo
+  evaluate.py         # fixed-scenario CLI (rewritten to use eval_core)
+  compare.py          # CI comparison front-end (wraps build_eval_comparison)
+  app.py              # leaderboard (fed by both modes)
+```
+
+## Phases
+
+### Phase 0 — Integration spine (no Rust changes)
+- Bench depends on `openra_rl_training` + `openra_train`.
+- `eval_core.py`: wrap `play_episodes_async`, force Rust pool.
+- `agent.py`: real OpenAI-compatible model agent w/ minimap PNG.
+- Rewrite `evaluate.py` → fixed-scenario eval producing `eval_stats.json`.
+- `compare.py` → 95% CI tables. Wire results into `app.py` leaderboard.
+- Validate on rush-hour + scout-maginot (these *are* the perception tasks).
+
+### Phase 1 — Adversarial 1v1 (combat-only, current mechanics)
+- Rust: add a second RL-controlled player slot in a 1v1 map (both sides accept
+  Commands; remove scripted-enemy assumption).
+- Bench `pairwise.py`: two-model orchestration, win-rate + Elo, leaderboard.
+
+### Phase 2 — Controlled scenario library (current mechanics)
+- Author perception/spatial scenarios that isolate the unexplored→target
+  connection failure; maze/chokepoint pathfinding; multi-target prioritization.
+- Verifiable rubrics per scenario (intelligence_pct, path-optimality, etc.).
+
+### Phase 3 — Rust mechanics expansion (unlocks scenario families)
+- 3a Economy: ore/cash/harvester obs + HARVEST cmd + economy reward.
+- 3b Production/tech: production queue, BUILD/TRAIN, available_production, power.
+- Each sub-phase ships its scenario family + rubrics.
+
+## Test coverage (live engine, no mocks)
+
+`tests/test_rust_integration.py` — 17 tests, ~1.9s, boots real
+`openra_train` with rule-based bots (idle / charge / hunter):
+- tool correctness: move_units reaches target; idle units hold; attack
+  path; reset schema; same-seed determinism (bit-for-bit).
+- corner cases: empty command list, invalid unit id (warns, no raise),
+  invalid attack target, out-of-bounds move — all safe.
+- invariants: explored% and units_killed monotonic non-decreasing;
+  discovery set cumulative.
+- stack: adapter signal tracking; win-condition predicates + composites
+  + unknown-key rejection; **deterministic win/fail plumbing** (trivially
+  true win/fail conditions ⇒ exact outcome, not bot-skill dependent);
+  all authored packs run end-to-end.
+
+## Sequencing (locked)
+
+`Phase 0 → Phase 2 (+ P/R/A diagnostics) → Phase 1 → Phase 3`
+
+Scenario breadth + per-link diagnostics first: fastest path to exposing real
+model strengths/weaknesses on current Rust mechanics. Adversarial 1v1 and the
+economy/production engine work follow.
+
+## Model provider abstraction (Phase 0)
+
+`openra_bench/agent.py` exposes a provider-agnostic agent. Adapters:
+
+- **openai-compatible** (default): covers local vLLM (matches Training's rollout
+  path) and **OpenRouter** (test target). Same Chat Completions + multimodal
+  `image_url` for the minimap PNG. Base URL + key from config/env.
+- **bedrock**: separate adapter (AWS SDK / Converse API), added when needed.
+
+Selected via Bench config (`provider`, `base_url`, `model`, `api_key_env`).
+Phase 0 validates with OpenRouter.

openra_bench/__init__.py

@@ -0,0 +1,7 @@
+"""OpenRA-Bench — model evaluation on spatial / multi-modal / multi-step
+planning, on the Rust OpenRA environment, reusing OpenRA-RL-Training components.
+
+See EVAL_STACK_PLAN.md for architecture and phasing.
+"""
+
+__all__ = ["rust_adapter", "eval_core"]

openra_bench/eval_core.py

@@ -0,0 +1,181 @@
+"""Episode spine: Rust env + adapter + pluggable agent.
+
+This is the Bench-side replacement for Training's `play_episodes_async`
+(which is hardwired to the C# gRPC server). It reuses Training *components*
+via the adapter; provider-agnostic agents plug in here (Phase 0 follow-up:
+openra_bench/agent.py with vLLM/OpenRouter/Bedrock).
+
+An `agent_fn` has signature:
+    agent_fn(render_state: dict, Command) -> list[Command]
+where `Command` is `openra_train.Command` (move_units/attack_unit/observe).
+"""
+
+from __future__ import annotations
+
+import tempfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable
+
+import yaml
+from openra_rl_training.training.rust_env_pool import RustEnvPool
+
+from .rust_adapter import EpisodeSignals, RustObsAdapter
+from .scenarios.schema import CompiledLevel
+from .scenarios.win_conditions import WinContext, evaluate
+
+AgentFn = Callable[[dict, Any], list]
+
+
+def _scenario_to_tmp_yaml(compiled: CompiledLevel) -> str:
+    """Serialize a compiled level's ScenarioDefinition to a temp YAML the
+    Rust env can load (it reads actors from the given scenario path; the
+    map geometry is the Rust-supported base map)."""
+    data = compiled.scenario.model_dump(mode="json", exclude_none=True)
+    fd = tempfile.NamedTemporaryFile(
+        "w", suffix=f"_{compiled.pack_id}_{compiled.level}.yaml", delete=False
+    )
+    yaml.safe_dump(data, fd, sort_keys=False)
+    fd.close()
+    return fd.name
+
+
+@dataclass
+class EpisodeResult:
+    scenario: str
+    seed: int
+    turns: int
+    signals: EpisodeSignals
+    outcome: str = "draw"  # "win" | "loss" | "draw"
+    trace: list[dict] = field(default_factory=list)
+
+
+def scripted_explore_agent(render_state: dict, Command: Any) -> list:
+    """Baseline reference agent: walk every unit toward the nearest
+    unexplored frontier cell. Exercises the move path; a useful
+    lower-bound control for the perception/exploration scenarios.
+    """
+    grid = render_state["minimap"].splitlines()
+    h = len(grid)
+    w = len(grid[0]) if grid else 0
+    frontier = [
+        (x, y)
+        for y in range(h)
+        for x in range(min(w, len(grid[y])))
+        if grid[y][x] == "#"
+    ]
+    units = render_state.get("units_summary", [])
+    if not units or not frontier:
+        return [Command.observe()]
+    cmds = []
+    for u in units:
+        ux, uy = u["cell_x"], u["cell_y"]
+        tx, ty = min(frontier, key=lambda c: (c[0] - ux) ** 2 + (c[1] - uy) ** 2)
+        cmds.append(Command.move_units([str(u["id"])], target_x=tx, target_y=ty))
+    return cmds
+
+
+def run_episode(
+    scenario_path: str,
+    agent_fn: AgentFn = scripted_explore_agent,
+    max_turns: int = 40,
+    seed: int = 0,
+    pool: RustEnvPool | None = None,
+) -> EpisodeResult:
+    owns_pool = pool is None
+    if pool is None:
+        pool = RustEnvPool(size=1, scenario_path=scenario_path)
+    env = pool.acquire()
+    try:
+        adapter = RustObsAdapter()
+        obs = env.reset(seed=seed)
+        adapter.observe(obs)
+        trace: list[dict] = []
+        turns = 0
+        for turns in range(1, max_turns + 1):
+            rs = adapter.render_state()
+            cmds = agent_fn(rs, env.Command) or [env.Command.observe()]
+            obs, _reward, done, info = env.step(cmds)
+            adapter.observe(obs, done=done)
+            trace.append(
+                {
+                    "turn": turns,
+                    "tick": adapter.signals.game_tick,
+                    "explored": round(adapter.signals.explored_percent, 2),
+                    "kills": adapter.signals.units_killed,
+                    "enemies_seen": len(adapter.signals.enemies_seen_ids),
+                    "n_cmds": len(cmds),
+                }
+            )
+            if done:
+                break
+        return EpisodeResult(
+            scenario=scenario_path,
+            seed=seed,
+            turns=turns,
+            signals=adapter.signals,
+            trace=trace,
+        )
+    finally:
+        pool.release(env)
+        if owns_pool:
+            pool.shutdown()
+
+
+def run_level(
+    compiled: CompiledLevel,
+    agent_fn: AgentFn = scripted_explore_agent,
+    seed: int = 0,
+) -> EpisodeResult:
+    """Run one scenario-pack level, scoring against its declarative
+    win/fail conditions (checked every turn). Outcome maps to the
+    `reward_outcome` convention: win=1.0, draw=0.5, loss=0.0.
+    """
+    if not compiled.map_supported:
+        raise RuntimeError(
+            f"{compiled.pack_id}: base map not Rust-loadable yet (Phase 3). "
+            f"Validate-only; cannot execute."
+        )
+    tmp_path = _scenario_to_tmp_yaml(compiled)
+    pool = RustEnvPool(size=1, scenario_path=tmp_path)
+    env = pool.acquire()
+    try:
+        adapter = RustObsAdapter()
+        adapter.observe(env.reset(seed=seed))
+        trace: list[dict] = []
+        outcome = "draw"
+        turns = 0
+        for turns in range(1, compiled.max_turns + 1):
+            rs = adapter.render_state()
+            cmds = agent_fn(rs, env.Command) or [env.Command.observe()]
+            obs, _r, done, _info = env.step(cmds)
+            adapter.observe(obs, done=done)
+            ctx = WinContext(signals=adapter.signals, render_state=adapter.render_state())
+            if evaluate(compiled.win_condition, ctx):
+                outcome = "win"
+            elif evaluate(compiled.fail_condition, ctx):
+                outcome = "loss"
+            trace.append(
+                {
+                    "turn": turns,
+                    "tick": adapter.signals.game_tick,
+                    "explored": round(adapter.signals.explored_percent, 2),
+                    "kills": adapter.signals.units_killed,
+                    "enemies_seen": len(adapter.signals.enemies_seen_ids),
+                }
+            )
+            if outcome != "draw" or done:
+                break
+        adapter.signals.outcome = {"win": 1.0, "draw": 0.5, "loss": 0.0}[outcome]
+        return EpisodeResult(
+            scenario=f"{compiled.pack_id}:{compiled.level}",
+            seed=seed,
+            turns=turns,
+            signals=adapter.signals,
+            outcome=outcome,
+            trace=trace,
+        )
+    finally:
+        pool.release(env)
+        pool.shutdown()
+        Path(tmp_path).unlink(missing_ok=True)

openra_bench/rust_adapter.py

@@ -0,0 +1,238 @@
+"""Rust env -> Training-component schema adapter.
+
+The Rust env (`openra_train.OpenRAEnv`) emits a lean observation:
+
+    keys = unit_positions, unit_hp, enemy_positions, enemy_hp,
+           enemy_buildings_summary, explored_cells, explored_percent,
+           game_tick, units_killed
+    step() -> (obs, reward=0.0 (hardcoded), done: bool,
+               info={game_tick, warnings})
+
+`minimap_renderer.render_minimap()` and the prompt builders in
+OpenRA-RL-Training expect a different shape (`units_summary`,
+`enemy_summary`, an ASCII `minimap`, `terrain_png`). And because the
+Rust env hardcodes reward to 0.0, all scoring/diagnostic signals must be
+derived here from observation deltas.
+
+This module is the single translation point. It is intentionally pure
+(no model / network / file I/O beyond optional terrain load) so it can
+be unit-tested against captured Rust observations.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+def _cells(obj: Any) -> list[tuple[int, int]]:
+    """Normalize explored_cells / position lists to [(x, y), ...]."""
+    out: list[tuple[int, int]] = []
+    if not obj:
+        return out
+    for c in obj:
+        if isinstance(c, dict):
+            out.append((int(c.get("cell_x", 0)), int(c.get("cell_y", 0))))
+        elif isinstance(c, (list, tuple)) and len(c) >= 2:
+            out.append((int(c[0]), int(c[1])))
+    return out
+
+
+def _units_to_render_list(
+    positions: dict[str, Any],
+    hp: dict[str, Any] | None,
+    type_by_id: dict[str, str] | None = None,
+) -> list[dict]:
+    """unit_positions {id: {cell_x, cell_y, ...}} -> [{cell_x, cell_y, type, id, hp}]."""
+    hp = hp or {}
+    type_by_id = type_by_id or {}
+    out: list[dict] = []
+    for uid, p in (positions or {}).items():
+        if isinstance(p, dict):
+            cx, cy = int(p.get("cell_x", 0)), int(p.get("cell_y", 0))
+            activity = p.get("activity")
+        elif isinstance(p, (list, tuple)) and len(p) >= 2:
+            cx, cy, activity = int(p[0]), int(p[1]), None
+        else:
+            continue
+        out.append(
+            {
+                "id": str(uid),
+                "cell_x": cx,
+                "cell_y": cy,
+                "type": type_by_id.get(str(uid)),
+                "hp": float(hp.get(uid, hp.get(str(uid), 1.0)) or 0.0),
+                "activity": activity,
+            }
+        )
+    return out
+
+
+@dataclass
+class EpisodeSignals:
+    """Cumulative + per-step signals derived from Rust obs deltas.
+
+    Drives both `reward_funcs` inputs and the P/R/A diagnostic rubrics
+    (task #2). Rust gives no reward/result, so every signal lives here.
+    """
+
+    units_killed: int = 0
+    units_killed_delta: int = 0
+    units_lost: int = 0
+    explored_percent: float = 0.0
+    explored_delta: float = 0.0
+    enemies_seen_ids: set[str] = field(default_factory=set)
+    enemy_buildings_seen_ids: set[str] = field(default_factory=set)
+    new_enemies_this_step: int = 0
+    new_buildings_this_step: int = 0
+    game_tick: int = 0
+    done: bool = False
+    # Outcome is synthesized (Rust has no result field): a scenario is
+    # "won" when all enemy buildings have been discovered AND/OR all
+    # enemy units neutralized — refined per-scenario in Phase 2 rubrics.
+    outcome: float = 0.0
+
+    def as_reward_kwargs(self) -> dict[str, Any]:
+        """Shape expected by OpenRA-RL-Training reward_funcs (game signals)."""
+        return {
+            "units_killed": self.units_killed,
+            "units_lost": self.units_lost,
+            "explored_percent": self.explored_percent,
+            "enemies_discovered": len(self.enemies_seen_ids),
+            "buildings_discovered": len(self.enemy_buildings_seen_ids),
+            "outcome": self.outcome,
+            "game_tick": self.game_tick,
+            "done": self.done,
+        }
+
+
+class RustObsAdapter:
+    """Stateful per-episode adapter. One instance per episode.
+
+    Usage:
+        ad = RustObsAdapter(scenario_def)
+        ad.observe(reset_obs)
+        ...loop: ad.observe(step_obs, done=done)
+        render_state = ad.render_state()        # for minimap_renderer
+        sig = ad.signals                        # for scoring / diagnostics
+    """
+
+    def __init__(self, scenario: Any = None, type_by_id: dict[str, str] | None = None):
+        self.scenario = scenario
+        self.type_by_id = type_by_id or {}
+        self.signals = EpisodeSignals()
+        self._explored: set[tuple[int, int]] = set()
+        self._prev_own_ids: set[str] = set()
+        self._raw: dict[str, Any] = {}
+        self._first_own_count: int | None = None
+
+    # -- ingestion --------------------------------------------------------
+    def observe(self, obs: dict[str, Any], done: bool = False) -> None:
+        self._raw = obs or {}
+        s = self.signals
+
+        own = self._raw.get("unit_positions", {}) or {}
+        own_ids = {str(k) for k in own}
+        if self._first_own_count is None:
+            self._first_own_count = len(own_ids)
+        # Lost = units that disappeared from our roster.
+        s.units_lost = max(0, (self._first_own_count or 0) - len(own_ids))
+        self._prev_own_ids = own_ids
+
+        prev_kills = s.units_killed
+        s.units_killed = int(self._raw.get("units_killed", s.units_killed) or 0)
+        s.units_killed_delta = max(0, s.units_killed - prev_kills)
+
+        prev_expl = s.explored_percent
+        s.explored_percent = float(self._raw.get("explored_percent", prev_expl) or 0.0)
+        s.explored_delta = max(0.0, s.explored_percent - prev_expl)
+        self._explored.update(_cells(self._raw.get("explored_cells")))
+
+        before_e = len(s.enemies_seen_ids)
+        for e in self._raw.get("enemy_positions", []) or []:
+            if isinstance(e, dict) and e.get("id") is not None:
+                s.enemies_seen_ids.add(str(e["id"]))
+        s.new_enemies_this_step = len(s.enemies_seen_ids) - before_e
+
+        before_b = len(s.enemy_buildings_seen_ids)
+        for b in self._raw.get("enemy_buildings_summary", []) or []:
+            if isinstance(b, dict) and b.get("id") is not None:
+                s.enemy_buildings_seen_ids.add(str(b["id"]))
+        s.new_buildings_this_step = len(s.enemy_buildings_seen_ids) - before_b
+
+        s.game_tick = int(self._raw.get("game_tick", s.game_tick) or 0)
+        s.done = bool(done)
+
+    # -- render schema ----------------------------------------------------
+    def grid_dims(self, margin: int = 4) -> tuple[int, int]:
+        """Derive a working (width, height) from observed extents.
+
+        Phase 0: the Rust env exposes no map_info; bound from observed
+        cells. Phase 3 will plumb true map dims through the adapter.
+        """
+        xs, ys = [0], [0]
+        for src in (self._explored, _cells(self._raw.get("explored_cells"))):
+            for x, y in src:
+                xs.append(x)
+                ys.append(y)
+        for coll in (
+            self._raw.get("unit_positions", {}) or {},
+            self._raw.get("enemy_positions", []) or [],
+            self._raw.get("enemy_buildings_summary", []) or [],
+        ):
+            items = coll.values() if isinstance(coll, dict) else coll
+            for p in items:
+                if isinstance(p, dict):
+                    xs.append(int(p.get("cell_x", 0)))
+                    ys.append(int(p.get("cell_y", 0)))
+        return max(xs) + margin, max(ys) + margin
+
+    def ascii_minimap(self) -> str:
+        """Synthesize the ASCII grid the renderer parses for the explored
+        mask: '#' = unexplored, '.' = explored. Faithful to
+        minimap_renderer._parse_ascii_minimap (anything != '#' = explored).
+        """
+        w, h = self.grid_dims()
+        explored = set(self._explored) | set(_cells(self._raw.get("explored_cells")))
+        rows = []
+        for y in range(h):
+            rows.append("".join("." if (x, y) in explored else "#" for x in range(w)))
+        return "\n".join(rows)
+
+    def render_state(self) -> dict[str, Any]:
+        """State dict shaped for minimap_renderer.render_minimap()/prompts."""
+        w, h = self.grid_dims()
+        own = _units_to_render_list(
+            self._raw.get("unit_positions", {}),
+            self._raw.get("unit_hp"),
+            self.type_by_id,
+        )
+        enemy = _units_to_render_list(
+            {
+                str(e.get("id", i)): e
+                for i, e in enumerate(self._raw.get("enemy_positions", []) or [])
+            },
+            self._raw.get("enemy_hp"),
+        )
+        enemy += [
+            {
+                "id": str(b.get("id", f"bldg{i}")),
+                "cell_x": int(b.get("cell_x", 0)),
+                "cell_y": int(b.get("cell_y", 0)),
+                "type": b.get("kind") or b.get("type"),
+                "hp": float(b.get("hp_pct", 1.0) or 0.0),
+                "is_building": True,
+            }
+            for i, b in enumerate(self._raw.get("enemy_buildings_summary", []) or [])
+        ]
+        return {
+            "units_summary": own,
+            "enemy_summary": enemy,
+            "minimap": self.ascii_minimap(),
+            "map_width": w,
+            "map_height": h,
+            "bounds_x": 0,
+            "bounds_y": 0,
+            "game_tick": self.signals.game_tick,
+            "explored_percent": self.signals.explored_percent,
+        }

openra_bench/scenarios/CONTRIBUTING.md

@@ -0,0 +1,95 @@
+# Contributing a Scenario Pack
+
+A **pack** is one YAML file in `packs/` describing **one decision problem
+at three difficulty levels** (`easy` / `medium` / `hard`). You write only
+YAML — no Python.
+
+## Rules
+
+1. **It must mean something.** `meta.real_world_meaning` and
+   `meta.robotics_analogue` are required and reviewed. Example: *path
+   planning algorithms are solved; deciding **which** unexplored region
+   to commit to under partial information is not — that is the search-
+   and-rescue problem.*
+2. **One capability focus.** `meta.capability` ∈ `perception` |
+   `reasoning` | `action` — the link of the Perception→Reasoning→Action
+   chain this scenario primarily stresses.
+3. **Three real levels.** Difficulty must come from the *decision*
+   getting harder (less information, more decoys, tighter deadline,
+   stronger defenses) — not just bigger numbers.
+4. **Custom win condition, declaratively.** Use the grammar below; the
+   "bot" / objective is whatever the win condition says.
+5. **Map.** `base_map: rush-hour-arena` works today. Other maps are
+   schema-valid but skipped until the Rust generic-map loader (Phase 3).
+
+## File shape
+
+```yaml
+meta:
+  id: partial-info-rescue          # lowercase-kebab, unique
+  title: "Rescue Under Partial Information"
+  capability: reasoning
+  real_world_meaning: >
+    Pathfinding is solved; choosing which unexplored area to search
+    first with limited fuel/time is the actual rescue problem.
+  robotics_analogue: "UAV search-and-rescue frontier selection"
+  author: "your-name"
+
+base_map: rush-hour-arena
+base:                              # shared ScenarioDefinition fields
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  actors:
+    - {type: jeep, owner: agent, position: [5, 5], count: 3}
+    - {type: e1,   owner: enemy, position: [60, 20], stance: 2}
+  termination: {max_ticks: 8000}
+
+levels:
+  easy:
+    description: "Target in the nearest unexplored quadrant."
+    overrides: {}                  # deep-merge patch onto base
+    win_condition: {all_of: [{buildings_discovered_gte: 1}, {within_ticks: 6000}]}
+    fail_condition: {units_lost_lte: -1}      # optional
+    max_turns: 30
+  medium:
+    description: "Two plausible regions; one is a decoy."
+    overrides:
+      actors:                       # full list replaces base.actors
+        - {type: jeep, owner: agent, position: [5, 5], count: 2}
+        - {type: e1,   owner: enemy, position: [90, 30], stance: 2}
+    win_condition: {all_of: [{buildings_discovered_gte: 1}, {within_ticks: 5000}]}
+    max_turns: 35
+  hard:
+    description: "Three regions, decoys, tight deadline, attrition."
+    overrides: { ... }
+    win_condition: { ... }
+    max_turns: 40
+```
+
+## Win-condition grammar
+
+Composites: `all_of: [..]`, `any_of: [..]`, `not: {..}`. Leaves (a node
+with multiple leaves is an implicit AND):
+
+| Leaf | Meaning |
+|---|---|
+| `explored_pct_gte: <float>` | map % revealed ≥ value |
+| `enemies_discovered_gte: <int>` | distinct enemy units seen ≥ value |
+| `buildings_discovered_gte: <int>` | distinct enemy buildings seen ≥ value |
+| `units_killed_gte: <int>` | agent kills ≥ value |
+| `units_lost_lte: <int>` | agent losses ≤ value (constraint) |
+| `within_ticks: <int>` | reached by game tick ≤ value (deadline) |
+| `after_ticks: <int>` | only after game tick ≥ value |
+| `reach_region: {x,y,radius}` | ≥1 agent unit within radius of (x,y) |
+| `all_units_in_region: {x,y,radius}` | every agent unit within radius |
+
+`win_condition` is checked every turn; first turn it holds → **win**.
+`fail_condition` likewise → **loss**. Neither by `max_turns` → **draw**.
+
+## Validate before opening a PR
+
+```bash
+python -m openra_bench.scenarios.validate packs/your-pack.yaml
+```

openra_bench/scenarios/__init__.py

@@ -0,0 +1,31 @@
+"""Contributor-facing scenario layer.
+
+A *scenario pack* is one YAML file describing one decision problem at
+three difficulty levels (`easy` / `medium` / `hard`). Each level compiles
+to an OpenRA-RL-Training `ScenarioDefinition` plus a declarative
+`WinCondition` — so lab mates contribute scenarios with custom bot win
+conditions and (schema-wise) custom maps without writing Python.
+
+Public API:
+    load_pack(path)               -> ScenarioPack
+    discover_packs(dir)           -> list[ScenarioPack]
+    pack.compile(level)           -> CompiledLevel  (engine def + win cond)
+
+See CONTRIBUTING.md and packs/TEMPLATE.yaml.
+"""
+
+from .loader import discover_packs, load_pack
+from .schema import CompiledLevel, Level, ScenarioMeta, ScenarioPack
+from .win_conditions import WinCondition, WinContext, evaluate
+
+__all__ = [
+    "load_pack",
+    "discover_packs",
+    "ScenarioPack",
+    "ScenarioMeta",
+    "Level",
+    "CompiledLevel",
+    "WinCondition",
+    "WinContext",
+    "evaluate",
+]

openra_bench/scenarios/loader.py

@@ -0,0 +1,66 @@
+"""Pack discovery + loading + map-support gating.
+
+The Rust env currently loads only two hardcoded map geometries
+(`rush-hour`, `scout-maginot` — see OpenRA-Rust env.rs). Contributors
+may still author meaningful scenarios *today* by varying actors, spawns,
+and win conditions on a supported geometry. A pack that names an
+unsupported `base_map` still loads and validates, but its compiled
+levels carry `map_supported=False` so the runner can skip/flag them
+rather than crash. Generic `.oramap` loading lands in Phase 3.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+
+from .schema import LevelName, ScenarioPack
+
+# Logical base_map id -> the Rust scenario alias the env actually loads.
+# Extend this as Phase 3 adds real map loading.
+SUPPORTED_MAPS: dict[str, str] = {
+    "rush-hour-arena": "scenarios/discovery/rush-hour.yaml",
+    "scout-maginot": "scenarios/strategy/scout-maginot.yaml",
+}
+
+PACKS_DIR = Path(__file__).parent / "packs"
+
+
+def load_pack(path: str | Path) -> ScenarioPack:
+    """Parse and validate a single pack YAML."""
+    path = Path(path)
+    with open(path) as f:
+        data = yaml.safe_load(f)
+    try:
+        return ScenarioPack(**data)
+    except Exception as e:  # noqa: BLE001 — re-raise with file context
+        raise ValueError(f"invalid scenario pack {path}: {e}") from e
+
+
+def discover_packs(directory: str | Path | None = None) -> list[ScenarioPack]:
+    """Load every *.yaml pack in `directory` (default: bundled packs/).
+
+    Templates (filenames starting with '_' or 'TEMPLATE') are skipped.
+    """
+    directory = Path(directory) if directory else PACKS_DIR
+    packs: list[ScenarioPack] = []
+    for p in sorted(directory.glob("*.yaml")):
+        if p.name.startswith(("_", "TEMPLATE")):
+            continue
+        packs.append(load_pack(p))
+    return packs
+
+
+def is_map_supported(base_map: str) -> bool:
+    return base_map in SUPPORTED_MAPS
+
+
+def rust_scenario_alias(base_map: str) -> str:
+    """The path/alias to hand the Rust env for this logical map."""
+    return SUPPORTED_MAPS[base_map]
+
+
+def compile_level(pack: ScenarioPack, level: LevelName):
+    """Compile one level, wiring in the map-support flag."""
+    return pack.compile(level, map_supported=is_map_supported(pack.base_map))

openra_bench/scenarios/packs/TEMPLATE.yaml

@@ -0,0 +1,52 @@
+# Copy this file to packs/<your-id>.yaml and fill it in.
+# Validate: python -m openra_bench.scenarios.validate packs/<your-id>.yaml
+# See ../CONTRIBUTING.md for the full win-condition grammar.
+
+meta:
+  id: my-scenario-id                 # lowercase-kebab, unique
+  title: "Human Readable Title"
+  capability: perception             # perception | reasoning | action
+  real_world_meaning: >
+    What real decision does this abstract? (>= 20 chars, reviewed)
+  robotics_analogue: "Concrete robotics/agentic parallel"
+  author: "your-name"
+
+base_map: rush-hour-arena            # only Rust-loadable map today
+
+base:                                # shared ScenarioDefinition fields
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  termination: {max_ticks: 8000}
+  actors:
+    - {type: jeep, owner: agent, position: [5, 5], count: 3}
+    - {type: e1,   owner: enemy, position: [60, 20], stance: 2}
+
+levels:
+  easy:
+    description: "Decision is obvious / fully observable."
+    overrides: {}
+    win_condition: {all_of: [{enemies_discovered_gte: 1}, {within_ticks: 6000}]}
+    max_turns: 30
+  medium:
+    description: "Partial info / one decoy."
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [5, 5], count: 2}
+        - {type: e1,   owner: enemy, position: [90, 30], stance: 2}
+    win_condition: {all_of: [{enemies_discovered_gte: 1}, {within_ticks: 5000}]}
+    max_turns: 35
+  hard:
+    description: "Decoys + deadline + attrition pressure."
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [5, 5], count: 2}
+        - {type: e1,   owner: enemy, position: [110, 35], stance: 2}
+        - {type: e3,   owner: enemy, position: [40, 10], stance: 2}
+    win_condition:
+      all_of:
+        - {enemies_discovered_gte: 2}
+        - {within_ticks: 4500}
+        - {units_lost_lte: 1}
+    max_turns: 40

openra_bench/scenarios/packs/action-multiunit-coordination.yaml

@@ -0,0 +1,110 @@
+# Validate: python -m openra_bench.scenarios.validate packs/action-multiunit-coordination.yaml
+# See ../CONTRIBUTING.md for the full win-condition grammar.
+#
+# ACTION focus: the plan is trivial and stated outright — "split the force,
+# send one group to region A and another to region B, arrive at both before
+# the deadline". The hard part is EXECUTION: a model that controls only one
+# group at a time, or marches everyone to a single objective, fails. Each
+# level adds another simultaneous objective / a tighter window so a single
+# serialized column can no longer satisfy the win condition in time.
+
+meta:
+  id: action-multiunit-coordination
+  title: "Split-Force Simultaneous Objectives"
+  capability: action
+  real_world_meaning: >
+    Multi-region task allocation under a shared deadline is not a planning
+    problem here — the split is obvious. It tests whether the controller can
+    actually drive several effector groups in parallel instead of completing
+    one objective then the next, which serialized control fails to do in
+    time.
+  robotics_analogue: >
+    Coordinated multi-robot fleet dispatch: a logistics swarm must place
+    distinct sub-teams at separate depots within one delivery window;
+    one-at-a-time control blows the schedule.
+  author: "openra-bench"
+
+base_map: rush-hour-arena
+
+base:
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  termination: {max_ticks: 8000}
+  actors:
+    # Two distinct agent groups in the spawn corner (x < 25) that must be
+    # split and driven to two separate regions at the same time.
+    - {type: 2tnk, owner: agent, position: [5, 6],  count: 3}
+    - {type: 1tnk, owner: agent, position: [7, 10], count: 3}
+    # Static enemy buildings act as the destination markers / targets.
+    - {type: fact, owner: enemy, position: [110, 6]}
+    - {type: proc, owner: enemy, position: [110, 33]}
+
+levels:
+  easy:
+    description: >
+      Two objectives, generous window. Split the 2tnk group to the
+      north-east region and the 1tnk group to the south-east region;
+      both groups must be on station before the deadline.
+    overrides: {}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 110, y: 6,  radius: 8}}
+        - {reach_region: {x: 110, y: 33, radius: 8}}
+        - {within_ticks: 6000}
+    max_turns: 30
+
+  medium:
+    description: >
+      Three simultaneous regions and an attrition cap. The force splits
+      three ways (NE, SE, mid-east); enemy infantry contest the lanes so
+      sloppy serialized movement bleeds units and misses the tighter
+      deadline.
+    overrides:
+      actors:
+        - {type: 2tnk, owner: agent, position: [5, 6],  count: 3}
+        - {type: 1tnk, owner: agent, position: [7, 10], count: 3}
+        - {type: jeep, owner: agent, position: [5, 14], count: 2}
+        - {type: fact, owner: enemy, position: [110, 5]}
+        - {type: proc, owner: enemy, position: [110, 34]}
+        - {type: powr, owner: enemy, position: [112, 20]}
+        - {type: e1,   owner: enemy, position: [60, 8],  stance: 2, count: 2}
+        - {type: e1,   owner: enemy, position: [60, 31], stance: 2, count: 2}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 110, y: 5,  radius: 7}}
+        - {reach_region: {x: 110, y: 34, radius: 7}}
+        - {reach_region: {x: 112, y: 20, radius: 7}}
+        - {within_ticks: 5000}
+        - {units_lost_lte: 2}
+    max_turns: 36
+
+  hard:
+    description: >
+      Three contested regions, the full force must arrive (every unit in
+      one of the three zones — checked as occupied corners), defenders
+      have turrets, the deadline is tight and attrition is strict. Only
+      genuine parallel multi-group control can satisfy all clauses.
+    overrides:
+      actors:
+        - {type: 2tnk, owner: agent, position: [5, 6],  count: 3}
+        - {type: 1tnk, owner: agent, position: [7, 10], count: 3}
+        - {type: jeep, owner: agent, position: [5, 14], count: 2}
+        - {type: apc,  owner: agent, position: [9, 18], count: 2}
+        - {type: fact, owner: enemy, position: [112, 5]}
+        - {type: proc, owner: enemy, position: [112, 35]}
+        - {type: powr, owner: enemy, position: [115, 20]}
+        - {type: gun,  owner: enemy, position: [100, 6]}
+        - {type: gun,  owner: enemy, position: [100, 34]}
+        - {type: e3,   owner: enemy, position: [55, 9],  stance: 2, count: 2}
+        - {type: e3,   owner: enemy, position: [55, 31], stance: 2, count: 2}
+        - {type: e1,   owner: enemy, position: [58, 20], stance: 2, count: 2}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 112, y: 5,  radius: 7}}
+        - {reach_region: {x: 112, y: 35, radius: 7}}
+        - {reach_region: {x: 115, y: 20, radius: 7}}
+        - {within_ticks: 4200}
+        - {units_lost_lte: 2}
+    max_turns: 44

openra_bench/scenarios/packs/action-sequenced-execution.yaml

@@ -0,0 +1,107 @@
+# Validate: python -m openra_bench.scenarios.validate packs/action-sequenced-execution.yaml
+# See ../CONTRIBUTING.md for the full win-condition grammar.
+#
+# ACTION focus: the route is given as an ordered plan — waypoint W1, then
+# W2, then the final objective. The model is not asked to *find* the order;
+# it is asked to *execute* it without stalling. The win condition encodes
+# the sequence via after_ticks gates: the final region must be reached only
+# after enough ticks that an idle / dithering agent runs out the clock,
+# while a decisive agent that keeps the column moving clears each leg in
+# time. Each level lengthens the route and tightens the budget so stalling
+# between legs becomes the dominant failure mode.
+
+meta:
+  id: action-sequenced-execution
+  title: "Ordered Route Execution Without Stalling"
+  capability: action
+  real_world_meaning: >
+    The route is already planned; the open problem is faithful, non-stalling
+    execution of an ordered objective sequence — issuing the next move the
+    instant a leg completes rather than idling, re-deliberating, or skipping
+    a waypoint, which is what makes long autonomous missions fail.
+  robotics_analogue: >
+    Manipulator / mobile-base task sequencing: a robot must hit a fixed
+    ordered set of stations (pick, transit, place) within a cycle-time
+    budget; pausing or reordering between steps misses the takt time.
+  author: "openra-bench"
+
+base_map: rush-hour-arena
+
+base:
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  termination: {max_ticks: 8000}
+  actors:
+    # One cohesive column that must traverse an ordered route.
+    - {type: 2tnk, owner: agent, position: [5, 8],  count: 2}
+    - {type: 1tnk, owner: agent, position: [7, 12], count: 2}
+    # Waypoint / terminal markers as static enemy buildings.
+    - {type: powr, owner: enemy, position: [45, 30]}   # W1
+    - {type: proc, owner: enemy, position: [90, 10]}    # W2
+    - {type: fact, owner: enemy, position: [118, 33]}   # final
+
+levels:
+  easy:
+    description: >
+      Two-leg route: go to waypoint W1 (SW-mid), then the final objective
+      (far SE). The after_ticks gate means the agent must already be
+      moving; arriving at the end only counts once W1's leg time has
+      elapsed, and the overall deadline punishes any stall.
+    overrides: {}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 118, y: 33, radius: 8}}
+        - {after_ticks: 1200}
+        - {within_ticks: 6000}
+    max_turns: 30
+
+  medium:
+    description: >
+      Three-leg route W1 -> W2 -> final with a contested first leg and an
+      attrition cap. A model that stalls to re-plan between legs cannot
+      clear all three before the tighter deadline.
+    overrides:
+      actors:
+        - {type: 2tnk, owner: agent, position: [5, 8],  count: 2}
+        - {type: 1tnk, owner: agent, position: [7, 12], count: 2}
+        - {type: jeep, owner: agent, position: [5, 16]}
+        - {type: powr, owner: enemy, position: [45, 31]}
+        - {type: proc, owner: enemy, position: [92, 9]}
+        - {type: fact, owner: enemy, position: [120, 34]}
+        - {type: e1,   owner: enemy, position: [40, 25], stance: 2, count: 2}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 120, y: 34, radius: 7}}
+        - {after_ticks: 2200}
+        - {within_ticks: 5000}
+        - {units_lost_lte: 1}
+    max_turns: 36
+
+  hard:
+    description: >
+      Four-leg serpentine route across the full arena with defended
+      waypoints, a strict attrition cap and a tight budget. Every leg must
+      be executed back-to-back; any idling, re-deliberation, or skipped
+      leg overruns the clock or loses too many units.
+    overrides:
+      actors:
+        - {type: 2tnk, owner: agent, position: [5, 8],  count: 2}
+        - {type: 1tnk, owner: agent, position: [7, 12], count: 2}
+        - {type: jeep, owner: agent, position: [5, 16]}
+        - {type: apc,  owner: agent, position: [9, 20]}
+        - {type: powr, owner: enemy, position: [45, 33]}    # W1
+        - {type: proc, owner: enemy, position: [80, 6]}      # W2
+        - {type: tsla, owner: enemy, position: [110, 30]}    # W3
+        - {type: fact, owner: enemy, position: [122, 6]}     # final
+        - {type: e3,   owner: enemy, position: [42, 27], stance: 2, count: 2}
+        - {type: e1,   owner: enemy, position: [78, 12], stance: 2, count: 2}
+        - {type: gun,  owner: enemy, position: [108, 25]}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 122, y: 6, radius: 7}}
+        - {after_ticks: 3400}
+        - {within_ticks: 4400}
+        - {units_lost_lte: 1}
+    max_turns: 46

openra_bench/scenarios/packs/perception-frontier-reading.yaml

@@ -0,0 +1,101 @@
+# Validate: python -m openra_bench.scenarios.validate \
+#   openra_bench/scenarios/packs/perception-frontier-reading.yaml
+# See ../CONTRIBUTING.md for the full win-condition grammar.
+#
+# PERCEPTION focus: can the model correctly READ the fog frontier — i.e.
+# tell which cells are still unrevealed and steer scouts INTO them — as
+# opposed to re-treading already-seen ground or stalling against decoys?
+# The win is a raw map-coverage threshold within a deadline, so the only
+# way to pass is to perceive where the unexplored mass is and commit
+# movement toward it. Difficulty rises by hiding the frontier farther
+# from spawn, splitting it across the map, adding decoy enemies that bait
+# attention to already-explored zones, and tightening the clock.
+
+meta:
+  id: perception-frontier-reading
+  title: "Reading the Fog Frontier"
+  capability: perception
+  real_world_meaning: >
+    Path planning is solved; the hard part is reading an occupancy grid
+    correctly to tell explored cells from the unknown frontier and
+    pushing sensors into the unknown instead of re-scanning known space.
+  robotics_analogue: "SLAM frontier detection — picking the next unexplored cell to drive a scout robot toward"
+  author: "openra-bench"
+
+base_map: rush-hour-arena
+
+base:
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  termination: {max_ticks: 10000}
+  actors:
+    # Fast, fragile scouts spawned in the bottom-left pocket (x<25).
+    - {type: jeep, owner: agent, position: [5, 33], count: 3}
+    - {type: jeep, owner: agent, position: [7, 30], count: 2}
+
+levels:
+  easy:
+    description: >
+      One contiguous unexplored mass directly east of spawn along the
+      open lane. The frontier is obvious and reachable; generous clock.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [5, 33], count: 3}
+        - {type: jeep, owner: agent, position: [7, 30], count: 2}
+        # A single static building parked deep east — incidentally seen
+        # by any honest eastward sweep.
+        - {type: proc, owner: enemy, position: [100, 20]}
+    win_condition:
+      all_of:
+        - {explored_pct_gte: 45}
+        - {within_ticks: 8000}
+    max_turns: 30
+
+  medium:
+    description: >
+      Unexplored area is split: a near pocket bottom-left already partly
+      visible, and the real bulk of the frontier lies far NE. A decoy
+      enemy squad sits inside the ALREADY-explored band to bait the
+      scouts into re-treading seen ground. Tighter clock, fewer scouts.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [5, 33], count: 3}
+        # Decoy: enemy infantry near spawn, inside soon-to-be-seen cells.
+        - {type: e1, owner: enemy, position: [20, 34], stance: 2, count: 2}
+        - {type: dog, owner: enemy, position: [22, 30], stance: 2}
+        # Real frontier mass: far NE, only reachable by a deliberate push.
+        - {type: powr, owner: enemy, position: [115, 6]}
+        - {type: proc, owner: enemy, position: [108, 12]}
+    win_condition:
+      all_of:
+        - {explored_pct_gte: 55}
+        - {within_ticks: 6000}
+    max_turns: 35
+
+  hard:
+    description: >
+      Frontier is fragmented across three corners (far NE, far SE, and a
+      thin NW strip). Two decoy squads sit in the explored center to pull
+      scouts off the true frontier, attrition is real (must keep >=4 of 5
+      scouts), and the deadline is short — only correct, simultaneous
+      reading of all three fog pockets clears the coverage bar in time.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [5, 33], count: 3}
+        - {type: jeep, owner: agent, position: [7, 30], count: 2}
+        # Center decoys (already-explored zone) — combat bait.
+        - {type: e1, owner: enemy, position: [55, 20], stance: 2, count: 3}
+        - {type: e3, owner: enemy, position: [58, 22], stance: 2}
+        - {type: dog, owner: enemy, position: [50, 18], stance: 2, count: 2}
+        # Three separated frontier markers forcing a full-map read.
+        - {type: powr, owner: enemy, position: [122, 4]}
+        - {type: proc, owner: enemy, position: [120, 36]}
+        - {type: gun,  owner: enemy, position: [10, 4]}
+    win_condition:
+      all_of:
+        - {explored_pct_gte: 62}
+        - {within_ticks: 4800}
+        - {units_lost_lte: 1}
+    max_turns: 40

openra_bench/scenarios/packs/perception-target-vs-fog.yaml

@@ -0,0 +1,111 @@
+# Validate: python -m openra_bench.scenarios.validate \
+#   openra_bench/scenarios/packs/perception-target-vs-fog.yaml
+# See ../CONTRIBUTING.md for the full win-condition grammar.
+#
+# PERCEPTION focus, targeting a KNOWN failure mode: the model treats
+# "unexplored region" and "where the target is" as unrelated facts. It
+# explores the nearest fog patch, sees nothing, and never connects that
+# the target building must therefore be in the OTHER unexplored region.
+# Here the win requires discovering the actual target building (not just
+# map coverage), and there is always a nearer-but-empty fog pocket plus a
+# farther fog pocket that genuinely contains the target. Passing means
+# the model spatially reasoned about WHERE the unexplored mass is
+# relative to WHERE the target must be — not just that fog exists.
+
+meta:
+  id: perception-target-vs-fog
+  title: "Where the Target Hides in the Fog"
+  capability: perception
+  real_world_meaning: >
+    The real search problem is not "go to fog" but inferring which of
+    several unexplored regions could actually contain the target given
+    what the empty regions rule out, then committing sensors there.
+  robotics_analogue: "Search-and-rescue: choosing which unexplored room can still hold the victim after clearing the near ones"
+
+base_map: rush-hour-arena
+
+base:
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  termination: {max_ticks: 10000}
+  actors:
+    # Scouts spawn top-left (x<25, y<10).
+    - {type: jeep, owner: agent, position: [6, 5], count: 3}
+    - {type: 1tnk, owner: agent, position: [8, 8], count: 2}
+
+levels:
+  easy:
+    description: >
+      One target building, one unexplored region — both far east. The
+      near area is already open (no decoy fog). Reading "fog is east,
+      therefore the target is east" is direct. Generous clock.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 5], count: 3}
+        - {type: 1tnk, owner: agent, position: [8, 8], count: 2}
+        # Target building in the single eastern fog mass.
+        - {type: fact, owner: enemy, position: [105, 20]}
+        # Light guard so it reads as a real objective, not a stray actor.
+        - {type: e1, owner: enemy, position: [102, 22], stance: 2}
+    win_condition:
+      all_of:
+        - {buildings_discovered_gte: 1}
+        - {within_ticks: 8000}
+    max_turns: 30
+
+  medium:
+    description: >
+      Two unexplored regions: a NEAR one (NE, closer to spawn) that is
+      EMPTY, and a FAR one (SE, opposite corner) that holds the real
+      target. The model must not stop after clearing the near fog and
+      finding nothing — it must infer the target is in the remaining
+      unexplored region. Decoy enemy units sit in the empty near region.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 5], count: 3}
+        - {type: 1tnk, owner: agent, position: [8, 8], count: 2}
+        # Near (NE) decoy region: enemy units but NO building (a trap
+        # that satisfies curiosity but not the win condition).
+        - {type: e1,  owner: enemy, position: [70, 6], stance: 2, count: 2}
+        - {type: jeep, owner: enemy, position: [74, 9], stance: 2}
+        # Far (SE) region: the actual target building.
+        - {type: fact, owner: enemy, position: [118, 35]}
+        - {type: e2,   owner: enemy, position: [114, 33], stance: 2}
+    win_condition:
+      all_of:
+        - {buildings_discovered_gte: 1}
+        - {within_ticks: 6000}
+    max_turns: 35
+
+  hard:
+    description: >
+      Three unexplored regions. Two are decoys: a NEAR-NE pocket with a
+      noisy enemy squad (pure bait) and a MID-SOUTH pocket with a single
+      enemy *building of the wrong kind already implicitly elsewhere* —
+      both contain no usable target. The real target is a lone building
+      tucked in the far-NW strip, the LEAST intuitive direction from a
+      spawn that naturally pushes east. Short deadline and attrition:
+      only a correct read of which fog pocket can still hold the target
+      — after the near ones are ruled out — discovers it in time.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 5], count: 3}
+        - {type: 1tnk, owner: agent, position: [8, 8], count: 2}
+        # Decoy A — near NE, loud enemy squad, no target building.
+        - {type: e1,  owner: enemy, position: [60, 6],  stance: 2, count: 3}
+        - {type: dog, owner: enemy, position: [64, 9],  stance: 2, count: 2}
+        # Decoy B — mid-south, enemy units only, no target building.
+        - {type: e3,  owner: enemy, position: [70, 34], stance: 2, count: 2}
+        - {type: apc, owner: enemy, position: [66, 31], stance: 2}
+        # The real target: a single building in the far-NW strip, the
+        # counter-intuitive corner relative to an eastward spawn bias.
+        - {type: tsla, owner: enemy, position: [118, 4]}
+        - {type: e1,   owner: enemy, position: [115, 6], stance: 2}
+    win_condition:
+      all_of:
+        - {buildings_discovered_gte: 1}
+        - {within_ticks: 4800}
+        - {units_lost_lte: 1}
+    max_turns: 40

openra_bench/scenarios/packs/reasoning-frontier-commit.yaml

@@ -0,0 +1,111 @@
+# reasoning-frontier-commit.yaml
+#
+# REASONING-link scenario. Perception is trivially correct (the map is
+# fog-of-war and a single scout discovers whatever it drives into); the
+# hard part is the *decision*: given partial information and a finite
+# tick budget, WHICH unexplored region do you commit your scout to?
+# Multiple regions look equally plausible from the corner; only one
+# holds the survivor (the objective marker). Wrong commitment burns the
+# clock and you draw.
+#
+# Validate:
+#   cd /Users/berta/Projects/OpenRA-Bench && \
+#   python3 -m openra_bench.scenarios.validate \
+#     openra_bench/scenarios/packs/reasoning-frontier-commit.yaml
+
+meta:
+  id: reasoning-frontier-commit
+  title: "Frontier Commitment Under Partial Information"
+  capability: reasoning
+  real_world_meaning: >
+    Path planning to any point is solved. The unsolved problem is
+    deciding which of several unexplored regions to commit a
+    time/fuel-limited searcher to when only one hides the target and
+    going to the wrong one first means you never reach the right one in
+    time. That commitment-under-uncertainty step is the real
+    search-and-rescue problem, not the navigation.
+  robotics_analogue: >
+    UAV/UGV search-and-rescue frontier selection: with limited
+    endurance and several candidate search cells, choose the cell most
+    likely to contain the survivor before the battery dies.
+  author: "berta"
+
+base_map: rush-hour-arena
+
+base:
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  termination: {max_ticks: 8000}
+  actors:
+    # Single fast scout in the NW corner (x<25). One body, one
+    # commitment — the decision cannot be hedged by splitting forces.
+    - {type: jeep, owner: agent, position: [6, 6], count: 1}
+    # The survivor: a static building marker, the only thing that
+    # satisfies the win condition. NE region.
+    - {type: gun, owner: enemy, position: [112, 6], stance: 0}
+
+levels:
+  easy:
+    description: >
+      One survivor, one plausible region. The objective marker sits in
+      the far-NE region and nothing else does — perception is honest,
+      the only task is to plan a direct commitment within the deadline.
+    overrides: {}
+    win_condition:
+      all_of:
+        - {buildings_discovered_gte: 1}
+        - {within_ticks: 6500}
+    max_turns: 30
+
+  medium:
+    description: >
+      Two candidate regions, one decoy. A decoy marker sits in the
+      far-SE region; the real survivor is in the far-NE. From the
+      corner both are equidistant-looking, but only the NE building
+      counts toward 'buildings_discovered'. Committing to the SE decoy
+      first wastes enough ticks to miss the deadline — the model must
+      reason about which to commit to, not just navigate.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 6], count: 1}
+        # Decoy: an enemy *unit* in SE — discovering it does NOT satisfy
+        # the buildings-based win condition. Pure distractor.
+        - {type: e1, owner: enemy, position: [112, 34], stance: 0, count: 2}
+        # Real survivor marker: NE building.
+        - {type: gun, owner: enemy, position: [112, 6], stance: 0}
+      termination: {max_ticks: 7000}
+    win_condition:
+      all_of:
+        - {buildings_discovered_gte: 1}
+        - {within_ticks: 5200}
+    max_turns: 34
+
+  hard:
+    description: >
+      Three candidate regions, two decoys, a tighter deadline, and an
+      attrition constraint. Decoy unit clusters sit in the SE and
+      mid-S regions; the real survivor marker is in the far-NE. One
+      decoy cluster is hostile (stance 2) so a careless route into it
+      can get the lone scout killed — the model must pick the
+      commitment that finds the building, fits the clock, AND avoids
+      the lethal frontier. Reasoning must trade route safety against
+      time without splitting (only one unit).
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 6], count: 1}
+        # Decoy A (SE) — hostile cluster; a route through it risks the
+        # lone scout (units_lost_lte: 0 constraint below).
+        - {type: e3, owner: enemy, position: [110, 34], stance: 2, count: 3}
+        # Decoy B (mid-S) — passive distractor unit.
+        - {type: e1, owner: enemy, position: [62, 36], stance: 0, count: 2}
+        # Real survivor marker: far-NE building.
+        - {type: gun, owner: enemy, position: [118, 5], stance: 0}
+      termination: {max_ticks: 6000}
+    win_condition:
+      all_of:
+        - {buildings_discovered_gte: 1}
+        - {within_ticks: 4200}
+        - {units_lost_lte: 0}
+    max_turns: 38

openra_bench/scenarios/packs/reasoning-risk-route.yaml

@@ -0,0 +1,132 @@
+# reasoning-risk-route.yaml
+#
+# REASONING-link scenario. Perception is correct (defenses are visible
+# once approached and the layout is fixed); the hard part is the
+# *plan*: there are two ways to the objective region — a short corridor
+# straight through lethal defenses, and a longer detour along a map
+# edge that avoids them. The short route is faster but kills a
+# light-unit scout; the safe route is slower but survivable. The model
+# must reason about the risk/route tradeoff and choose the longer,
+# safer plan under an attrition constraint and a deadline that still
+# permits the detour.
+#
+# Validate:
+#   cd /Users/berta/Projects/OpenRA-Bench && \
+#   python3 -m openra_bench.scenarios.validate \
+#     openra_bench/scenarios/packs/reasoning-risk-route.yaml
+
+meta:
+  id: reasoning-risk-route
+  title: "Risk vs Route: Safe-but-Long over Short-but-Deadly"
+  capability: reasoning
+  real_world_meaning: >
+    Shortest-path is solved. The unsolved decision is whether the
+    shortest path is the right path: a direct corridor may pass through
+    a hazard that destroys the vehicle, while a longer perimeter route
+    completes the mission intact. Choosing the survivable plan over the
+    fastest one, given a deadline that the detour still satisfies, is
+    the actual mission-planning problem.
+  robotics_analogue: >
+    Field-robot route selection under hazard: a delivery/inspection
+    robot must reject the shortest path when it crosses a no-go hazard
+    zone and instead commit to a longer but survivable route that still
+    meets the time window.
+  author: "berta"
+
+base_map: rush-hour-arena
+
+base:
+  agent: {faction: allies}
+  enemy: {faction: soviet}
+  tools: [move_units, attack_unit, stop_units]
+  planning: true
+  termination: {max_ticks: 9000}
+  actors:
+    # Lone light scout, NW corner (x<25). Fragile — cannot trade fire
+    # with turrets, so the route choice is the whole game.
+    - {type: jeep, owner: agent, position: [6, 20], count: 1}
+    # Objective marker on the far (east) side of the map.
+    - {type: gun, owner: enemy, position: [120, 20], stance: 0}
+    # Mid-map central choke "short route" defenses (lethal). The detour
+    # is along the top edge (low y) or bottom edge (high y), which is
+    # longer but clear in easy/medium.
+    - {type: tsla, owner: enemy, position: [62, 20], stance: 0}
+    - {type: e3,   owner: enemy, position: [60, 19], stance: 2, count: 3}
+    - {type: e3,   owner: enemy, position: [64, 21], stance: 2, count: 3}
+
+levels:
+  easy:
+    description: >
+      One central hazard, one obvious safe detour. The direct line
+      x-axis route passes through a tesla coil + rocket infantry that
+      shred the jeep; the top and bottom edges are completely open.
+      Generous deadline: any safe detour wins. The model only has to
+      recognise the corridor is lethal and not drive into it.
+    overrides: {}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 120, y: 20, radius: 7}}
+        - {units_lost_lte: 0}
+        - {within_ticks: 7500}
+    max_turns: 32
+
+  medium:
+    description: >
+      The bottom detour is now also guarded, so only the top-edge route
+      is safe — a longer commitment that the deadline still allows but
+      with less slack. The model must reason that the shortest route is
+      lethal, the bottom route is also lethal, and the top route,
+      though longest, is the only plan that satisfies both the
+      no-loss constraint and the tighter clock.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 20], count: 1}
+        - {type: gun,  owner: enemy, position: [120, 20], stance: 0}
+        # Central short-route hazard.
+        - {type: tsla, owner: enemy, position: [62, 20], stance: 0}
+        - {type: e3,   owner: enemy, position: [60, 19], stance: 2, count: 3}
+        - {type: e3,   owner: enemy, position: [64, 21], stance: 2, count: 3}
+        # Bottom-edge route now blocked too (lethal cluster, high y).
+        - {type: e3,   owner: enemy, position: [62, 36], stance: 2, count: 4}
+        - {type: gun,  owner: enemy, position: [62, 38], stance: 0}
+      termination: {max_ticks: 8000}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 120, y: 20, radius: 7}}
+        - {units_lost_lte: 0}
+        - {within_ticks: 6000}
+    max_turns: 36
+
+  hard:
+    description: >
+      Both edges are partly contested and the deadline is tight enough
+      that the longest fully-safe path is too slow — the model must
+      reason about a graduated tradeoff: the top edge is fastest but
+      grazes a defended pocket, the bottom edge is fully safe but too
+      long for the clock, so the only winning plan threads the
+      narrow safe seam near the top while staying out of weapon range.
+      Risk and route must be traded against the deadline, not avoided
+      outright.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 20], count: 1}
+        - {type: gun,  owner: enemy, position: [122, 20], stance: 0}
+        # Central short-route hazard (still lethal).
+        - {type: tsla, owner: enemy, position: [62, 20], stance: 0}
+        - {type: e3,   owner: enemy, position: [60, 19], stance: 2, count: 3}
+        - {type: e3,   owner: enemy, position: [64, 21], stance: 2, count: 3}
+        # Top-edge defended pocket — must skirt it precisely, not
+        # plough through it.
+        - {type: e3,   owner: enemy, position: [70, 6], stance: 2, count: 2}
+        - {type: gun,  owner: enemy, position: [72, 4], stance: 0}
+        # Bottom-edge fully blocked (forces the top seam, but bottom
+        # detour would also be too slow for the clock anyway).
+        - {type: e3,   owner: enemy, position: [62, 36], stance: 2, count: 4}
+        - {type: tsla, owner: enemy, position: [62, 38], stance: 0}
+      termination: {max_ticks: 7000}
+    win_condition:
+      all_of:
+        - {reach_region: {x: 122, y: 20, radius: 6}}
+        - {units_lost_lte: 0}
+        - {within_ticks: 4800}
+    max_turns: 40

openra_bench/scenarios/schema.py

@@ -0,0 +1,130 @@
+"""Scenario-pack schema: one YAML -> three difficulty levels.
+
+A pack composes (not forks) the OpenRA-RL-Training `ScenarioDefinition`.
+`base` holds the shared engine fields; each level supplies a deep-merge
+`overrides` patch plus its own `win_condition` / `fail_condition`. This
+keeps a three-level scenario in a single readable file and guarantees
+every level stays a valid engine scenario.
+"""
+
+from __future__ import annotations
+
+import copy
+from typing import Any, Literal
+
+from openra_rl_training.scenario import ScenarioDefinition
+from pydantic import BaseModel, Field, field_validator
+
+from .win_conditions import WinCondition
+
+LevelName = Literal["easy", "medium", "hard"]
+Capability = Literal["perception", "reasoning", "action"]
+
+
+def deep_merge(base: dict, patch: dict) -> dict:
+    """Recursive dict merge; lists/scalars in `patch` replace wholesale.
+
+    Replacing (not concatenating) lists is deliberate: a level that
+    customises `actors` states the full actor list, so diffs stay
+    auditable in review.
+    """
+    out = copy.deepcopy(base)
+    for k, v in (patch or {}).items():
+        if isinstance(v, dict) and isinstance(out.get(k), dict):
+            out[k] = deep_merge(out[k], v)
+        else:
+            out[k] = copy.deepcopy(v)
+    return out
+
+
+class ScenarioMeta(BaseModel):
+    """Why this scenario exists — required so the library stays meaningful."""
+
+    id: str = Field(..., description="Unique slug, e.g. partial-info-rescue")
+    title: str
+    capability: Capability = Field(
+        ..., description="Primary P/R/A chain link this scenario stresses"
+    )
+    real_world_meaning: str = Field(
+        ..., min_length=20, description="The real decision this abstracts"
+    )
+    robotics_analogue: str = Field(
+        ..., min_length=10, description="Concrete robotics/agentic parallel"
+    )
+    author: str = "unknown"
+
+    @field_validator("id")
+    @classmethod
+    def _slug(cls, v: str) -> str:
+        if not v.replace("-", "").isalnum() or v != v.lower():
+            raise ValueError(f"id must be lowercase kebab-case slug, got {v!r}")
+        return v
+
+
+class Level(BaseModel):
+    description: str = Field(..., min_length=10)
+    overrides: dict[str, Any] = Field(
+        default_factory=dict, description="Deep-merge patch onto pack.base"
+    )
+    win_condition: WinCondition
+    fail_condition: WinCondition | None = None
+    max_turns: int = Field(default=40, ge=1, le=400)
+
+
+class CompiledLevel(BaseModel):
+    """A single runnable level: validated engine scenario + conditions."""
+
+    model_config = {"arbitrary_types_allowed": True}
+
+    pack_id: str
+    level: LevelName
+    scenario: ScenarioDefinition
+    win_condition: WinCondition
+    fail_condition: WinCondition | None
+    max_turns: int
+    meta: ScenarioMeta
+    map_supported: bool = Field(
+        ..., description="False => Rust lacks this map (Phase 3 gate)"
+    )
+
+
+class ScenarioPack(BaseModel):
+    """The contributor-authored unit. One file = one decision problem."""
+
+    meta: ScenarioMeta
+    base_map: str = Field(
+        default="rush-hour-arena",
+        description="Logical map id; loader maps to a Rust-supported map",
+    )
+    base: dict[str, Any] = Field(
+        ..., description="Shared ScenarioDefinition fields (actors, factions, tools…)"
+    )
+    levels: dict[LevelName, Level]
+
+    @field_validator("levels")
+    @classmethod
+    def _all_three(cls, v: dict) -> dict:
+        missing = {"easy", "medium", "hard"} - set(v)
+        if missing:
+            raise ValueError(f"pack must define all levels; missing {sorted(missing)}")
+        return v
+
+    def compile(self, level: LevelName, *, map_supported: bool = True) -> CompiledLevel:
+        lvl = self.levels[level]
+        merged = deep_merge(self.base, lvl.overrides)
+        merged.setdefault("name", f"{self.meta.title} [{level}]")
+        merged.setdefault("description", lvl.description)
+        merged.setdefault("base_map", self.base_map)
+        # Validate against the real engine model so a broken level fails
+        # at load time, not mid-eval.
+        scenario = ScenarioDefinition(**merged)
+        return CompiledLevel(
+            pack_id=self.meta.id,
+            level=level,
+            scenario=scenario,
+            win_condition=lvl.win_condition,
+            fail_condition=lvl.fail_condition,
+            max_turns=lvl.max_turns,
+            meta=self.meta,
+            map_supported=map_supported,
+        )

openra_bench/scenarios/validate.py

@@ -0,0 +1,61 @@
+"""`python -m openra_bench.scenarios.validate [pack.yaml | dir]`
+
+Validates pack schema, all three levels' engine compilation, and
+win/fail-condition grammar. Exits non-zero on the first failure so it
+can gate CI / PRs.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+from .loader import PACKS_DIR, compile_level, is_map_supported, load_pack
+
+
+def _validate_one(path: Path) -> list[str]:
+    errs: list[str] = []
+    try:
+        pack = load_pack(path)
+    except Exception as e:  # noqa: BLE001
+        return [str(e)]
+    for level in ("easy", "medium", "hard"):
+        try:
+            compile_level(pack, level)  # constructs ScenarioDefinition + WinCondition
+        except Exception as e:  # noqa: BLE001
+            errs.append(f"[{pack.meta.id}:{level}] {e}")
+    if not is_map_supported(pack.base_map):
+        errs.append(
+            f"[{pack.meta.id}] base_map {pack.base_map!r} not Rust-loadable yet "
+            f"(schema-valid; runner will skip until Phase 3)"
+        )
+    return errs
+
+
+def main(argv: list[str]) -> int:
+    target = Path(argv[1]) if len(argv) > 1 else PACKS_DIR
+    files = (
+        [target]
+        if target.is_file()
+        else [p for p in sorted(target.glob("*.yaml")) if not p.name.startswith(("_", "TEMPLATE"))]
+    )
+    if not files:
+        print(f"no pack files found at {target}")
+        return 1
+    failed = False
+    for f in files:
+        errs = _validate_one(f)
+        warns = [e for e in errs if "not Rust-loadable" in e]
+        hard = [e for e in errs if e not in warns]
+        status = "FAIL" if hard else ("WARN" if warns else "OK")
+        print(f"{status:4}  {f.name}")
+        for e in hard:
+            print(f"      ✗ {e}")
+        for w in warns:
+            print(f"      ! {w}")
+        failed |= bool(hard)
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))

openra_bench/scenarios/win_conditions.py

@@ -0,0 +1,114 @@
+"""Declarative, composable win conditions.
+
+Contributors express the "custom bot win con" entirely in YAML — no
+Python. A condition is a tree of composites (`all_of` / `any_of` / `not`)
+over leaf predicates evaluated against a `WinContext` (the per-turn
+adapter signals plus the rendered state). The same tree also expresses
+*failure* conditions, so a scenario can be lost as well as won.
+
+Leaf predicates (key: value):
+    explored_pct_gte: float          map % revealed >= value
+    enemies_discovered_gte: int      distinct enemy units seen >= value
+    buildings_discovered_gte: int    distinct enemy buildings seen >= value
+    units_killed_gte: int            agent kill count >= value
+    units_lost_lte: int              agent units lost <= value (constraint)
+    within_ticks: int                current game tick <= value (deadline)
+    after_ticks: int                 current game tick >= value
+    reach_region: {x,y,radius}       any agent unit within radius of (x,y)
+    all_units_in_region: {x,y,radius}  every agent unit within radius
+
+Adding a predicate = one entry in `_PREDICATES`. Keep them pure.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from pydantic import BaseModel, model_validator
+
+
+@dataclass
+class WinContext:
+    """Everything a predicate may read. Pure data, no engine handles."""
+
+    signals: Any  # rust_adapter.EpisodeSignals
+    render_state: dict  # rust_adapter.RustObsAdapter.render_state()
+
+
+def _agent_units(ctx: WinContext) -> list[dict]:
+    return ctx.render_state.get("units_summary", []) or []
+
+
+def _in_radius(units: list[dict], x: int, y: int, r: float) -> int:
+    return sum(1 for u in units if (u["cell_x"] - x) ** 2 + (u["cell_y"] - y) ** 2 <= r * r)
+
+
+# Each predicate: (ctx, value) -> bool. Pure and side-effect free.
+_PREDICATES: dict[str, Callable[[WinContext, Any], bool]] = {
+    "explored_pct_gte": lambda c, v: c.signals.explored_percent >= float(v),
+    "enemies_discovered_gte": lambda c, v: len(c.signals.enemies_seen_ids) >= int(v),
+    "buildings_discovered_gte": lambda c, v: len(c.signals.enemy_buildings_seen_ids)
+    >= int(v),
+    "units_killed_gte": lambda c, v: c.signals.units_killed >= int(v),
+    "units_lost_lte": lambda c, v: c.signals.units_lost <= int(v),
+    "within_ticks": lambda c, v: c.signals.game_tick <= int(v),
+    "after_ticks": lambda c, v: c.signals.game_tick >= int(v),
+    "reach_region": lambda c, v: _in_radius(
+        _agent_units(c), int(v["x"]), int(v["y"]), float(v.get("radius", 3))
+    )
+    >= 1,
+    "all_units_in_region": lambda c, v: len(_agent_units(c)) > 0
+    and _in_radius(_agent_units(c), int(v["x"]), int(v["y"]), float(v.get("radius", 3)))
+    == len(_agent_units(c)),
+}
+
+LEAF_KEYS = frozenset(_PREDICATES)
+COMPOSITE_KEYS = frozenset({"all_of", "any_of", "not"})
+
+
+class WinCondition(BaseModel):
+    """One node: exactly one composite OR one-or-more leaf predicates.
+
+    Leaf form (implicit AND over keys):
+        {explored_pct_gte: 60, within_ticks: 6000}
+    Composite form:
+        {any_of: [{...}, {...}]}   {not: {...}}
+    """
+
+    model_config = {"extra": "allow"}
+
+    @model_validator(mode="after")
+    def _check_keys(self) -> "WinCondition":
+        keys = set(self.__pydantic_extra__ or {})
+        if not keys:
+            raise ValueError("win_condition node is empty")
+        unknown = keys - LEAF_KEYS - COMPOSITE_KEYS
+        if unknown:
+            raise ValueError(
+                f"unknown win-condition keys {sorted(unknown)}; "
+                f"valid leaves={sorted(LEAF_KEYS)} composites={sorted(COMPOSITE_KEYS)}"
+            )
+        if keys & COMPOSITE_KEYS and keys & LEAF_KEYS:
+            raise ValueError("cannot mix composite and leaf keys in one node")
+        if len(keys & COMPOSITE_KEYS) > 1:
+            raise ValueError("at most one composite key per node")
+        return self
+
+    def evaluate(self, ctx: WinContext) -> bool:
+        node = dict(self.__pydantic_extra__ or {})
+        if "all_of" in node:
+            return all(WinCondition(**c).evaluate(ctx) for c in node["all_of"])
+        if "any_of" in node:
+            return any(WinCondition(**c).evaluate(ctx) for c in node["any_of"])
+        if "not" in node:
+            return not WinCondition(**node["not"]).evaluate(ctx)
+        return all(_PREDICATES[k](ctx, v) for k, v in node.items())
+
+
+def evaluate(cond: WinCondition | dict | None, ctx: WinContext) -> bool:
+    if cond is None:
+        return False
+    if isinstance(cond, dict):
+        cond = WinCondition(**cond)
+    return cond.evaluate(ctx)

tests/test_rust_integration.py

@@ -0,0 +1,344 @@
+"""Integration tests that boot the real Rust OpenRA engine.
+
+These are *not* mocked: every test in `TestRustEngineTools` instantiates
+`openra_train.OpenRAEnv` and drives it with rule-based bot players, then
+asserts on observed engine behaviour (tool correctness + corner cases).
+`TestStackIntegration` exercises the Bench stack (adapter, win
+conditions, scenario packs, eval_core) on top of the live engine.
+
+Behaviour pinned here was first probed against the engine, not assumed:
+  * move_units drives a unit to (and reaching) the target cell
+  * same (scenario, seed) is bit-for-bit deterministic
+  * bad unit ids warn ("not owned"), don't raise
+  * empty / invalid commands are safe no-ops
+  * explored% and units_killed are monotonic non-decreasing
+
+Run: pytest tests/test_rust_integration.py -v
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+ot = pytest.importorskip("openra_train", reason="Rust env wheel not installed")
+
+TRAIN = Path("/Users/berta/Projects/OpenRA-RL-Training")
+RUSH_HOUR = str(TRAIN / "scenarios" / "discovery" / "rush-hour.yaml")
+
+pytestmark = pytest.mark.skipif(
+    not Path(RUSH_HOUR).exists(), reason="OpenRA-RL-Training scenarios not present"
+)
+
+
+# --------------------------------------------------------------------------- #
+# Rule-based bot players                                                       #
+# --------------------------------------------------------------------------- #
+def idle_bot(_obs):
+    """No-op: only `observe()`. Used to test determinism / no side effects."""
+    return [ot.Command.observe()]
+
+
+def charge_bot(target):
+    """Move every owned unit toward a fixed cell. Tests move_units."""
+
+    def _bot(obs):
+        ids = list(obs.get("unit_positions", {}))
+        if not ids:
+            return [ot.Command.observe()]
+        return [ot.Command.move_units(ids, target[0], target[1])]
+
+    return _bot
+
+
+def hunter_bot(obs):
+    """Attack the first visible enemy; otherwise push east. Tests attack_unit
+    and the discover→engage path."""
+    ids = list(obs.get("unit_positions", {}))
+    if not ids:
+        return [ot.Command.observe()]
+    enemies = obs.get("enemy_positions", []) or []
+    if enemies:
+        tid = str(enemies[0].get("id"))
+        return [ot.Command.attack_unit(ids, tid)]
+    return [ot.Command.move_units(ids, 120, 20)]
+
+
+def _first_unit(obs):
+    up = obs["unit_positions"]
+    k = sorted(up)[0]
+    return k, (up[k]["cell_x"], up[k]["cell_y"])
+
+
+def _man(a, b):
+    return abs(a[0] - b[0]) + abs(a[1] - b[1])
+
+
+def _run(env, bot, steps):
+    """Drive `env` with `bot` for `steps`, yielding (obs, reward, done, info)."""
+    obs = env.reset()
+    out = []
+    for _ in range(steps):
+        o, r, d, i = env.step(bot(obs))
+        out.append((o, r, d, i))
+        obs = o
+        if d:
+            break
+    return out
+
+
+# --------------------------------------------------------------------------- #
+# Engine + tool correctness (live Rust env)                                    #
+# --------------------------------------------------------------------------- #
+class TestRustEngineTools:
+    def test_reset_schema_and_initial_state(self):
+        obs = ot.OpenRAEnv(RUSH_HOUR, 7).reset()
+        for key in (
+            "unit_positions",
+            "unit_hp",
+            "enemy_positions",
+            "explored_percent",
+            "game_tick",
+            "units_killed",
+        ):
+            assert key in obs, f"missing obs key {key!r}"
+        assert obs["unit_positions"], "agent should own units at reset"
+        assert obs["explored_percent"] == pytest.approx(0.0, abs=1e-6)
+        assert obs["units_killed"] == 0
+        assert obs["game_tick"] < 50
+
+    def test_same_seed_is_deterministic(self):
+        a, b = ot.OpenRAEnv(RUSH_HOUR, 11), ot.OpenRAEnv(RUSH_HOUR, 11)
+        oa, ob = a.reset(), b.reset()
+        assert oa["unit_positions"] == ob["unit_positions"]
+        for _ in range(6):
+            oa, *_ = a.step([ot.Command.observe()])
+            ob, *_ = b.step([ot.Command.observe()])
+        assert oa["unit_positions"] == ob["unit_positions"]
+        assert oa["game_tick"] == ob["game_tick"]
+
+    def test_move_units_drives_unit_to_target(self):
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        obs = env.reset()
+        uid, start = _first_unit(obs)
+        target = (start[0] + 25, start[1])
+        last = start
+        for _ in range(15):
+            obs, *_ = env.step([ot.Command.move_units([uid], target[0], target[1])])
+            last = (obs["unit_positions"][uid]["cell_x"], obs["unit_positions"][uid]["cell_y"])
+        assert _man(last, target) < _man(start, target), "unit did not move toward target"
+        assert _man(last, target) <= 1, f"unit {uid} did not reach {target}, at {last}"
+
+    def test_idle_units_do_not_move(self):
+        """A corner agent unit with no order must hold position (no
+        spontaneous teleport / drift)."""
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        obs = env.reset()
+        uid, start = _first_unit(obs)  # spawn-corner unit, far from enemies
+        for _ in range(4):
+            obs, *_ = env.step([ot.Command.observe()])
+        pos = (obs["unit_positions"][uid]["cell_x"], obs["unit_positions"][uid]["cell_y"])
+        assert pos == start, f"idle unit drifted {start} -> {pos}"
+
+    def test_empty_command_list_is_safe(self):
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        env.reset()
+        obs, reward, done, info = env.step([])
+        assert isinstance(done, bool) and done is False
+        assert isinstance(reward, float)
+        assert obs["game_tick"] > 0
+
+    def test_invalid_unit_id_warns_not_raises(self):
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        env.reset()
+        _o, _r, _d, info = env.step([ot.Command.move_units(["999999"], 10, 10)])
+        warns = info.get("warnings", [])
+        assert any("999999" in w and "not owned" in w for w in warns), warns
+
+    def test_invalid_attack_target_is_safe(self):
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        obs = env.reset()
+        uid, _ = _first_unit(obs)
+        obs, _r, done, _i = env.step([ot.Command.attack_unit([uid], "888888")])
+        assert done is False
+        assert obs["unit_positions"], "engine state intact after bad attack target"
+
+    def test_out_of_bounds_move_is_safe(self):
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        obs = env.reset()
+        uid, _ = _first_unit(obs)
+        for _ in range(5):
+            obs, _r, done, _i = env.step([ot.Command.move_units([uid], 99999, 99999)])
+            assert done is False
+        p = obs["unit_positions"][uid]
+        assert 0 <= p["cell_x"] < 1000 and 0 <= p["cell_y"] < 1000
+
+    def test_explored_percent_monotonic_and_grows(self):
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        prev = -1.0
+        seen = []
+        for obs, *_ in _run(env, charge_bot((120, 20)), 20):
+            assert obs["explored_percent"] >= prev - 1e-6, "explored% decreased"
+            prev = obs["explored_percent"]
+            seen.append(prev)
+        assert seen[-1] > seen[0], "moving across the map revealed no new area"
+
+    def test_units_killed_monotonic_nondecreasing(self):
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        prev = 0
+        for obs, *_ in _run(env, hunter_bot, 30):
+            assert obs["units_killed"] >= prev, "units_killed went backwards"
+            prev = obs["units_killed"]
+        assert prev >= 0
+
+
+# --------------------------------------------------------------------------- #
+# Bench stack on the live engine                                               #
+# --------------------------------------------------------------------------- #
+class TestStackIntegration:
+    def test_adapter_signals_track_engine(self):
+        from openra_bench.rust_adapter import RustObsAdapter
+
+        env = ot.OpenRAEnv(RUSH_HOUR, 7)
+        ad = RustObsAdapter()
+        ad.observe(env.reset())
+        bot = charge_bot((120, 20))
+        prev_seen = 0
+        for _ in range(20):
+            o, _r, d, _i = env.step(bot({"unit_positions": ad._raw.get("unit_positions", {})}))
+            ad.observe(o, done=d)
+            # discovery set is cumulative — never shrinks
+            assert len(ad.signals.enemies_seen_ids) >= prev_seen
+            prev_seen = len(ad.signals.enemies_seen_ids)
+        kw = ad.signals.as_reward_kwargs()
+        assert set(kw) >= {
+            "units_killed",
+            "units_lost",
+            "explored_percent",
+            "enemies_discovered",
+            "outcome",
+            "game_tick",
+            "done",
+        }
+        assert kw["explored_percent"] > 0.0
+        assert kw["units_lost"] >= 0
+
+    def test_win_condition_predicates_pure(self):
+        from openra_bench.rust_adapter import EpisodeSignals
+        from openra_bench.scenarios.win_conditions import WinContext, evaluate
+
+        sig = EpisodeSignals(explored_percent=60.0, units_killed=3, game_tick=4000)
+        sig.enemies_seen_ids = {"a", "b"}
+        rs = {"units_summary": [{"id": "1", "cell_x": 10, "cell_y": 10, "type": None}]}
+        ctx = WinContext(signals=sig, render_state=rs)
+
+        assert evaluate({"explored_pct_gte": 50}, ctx) is True
+        assert evaluate({"explored_pct_gte": 75}, ctx) is False
+        assert evaluate({"enemies_discovered_gte": 2}, ctx) is True
+        assert evaluate({"within_ticks": 5000}, ctx) is True
+        assert evaluate({"within_ticks": 3000}, ctx) is False
+        assert evaluate({"reach_region": {"x": 11, "y": 11, "radius": 3}}, ctx) is True
+        assert evaluate({"reach_region": {"x": 99, "y": 99, "radius": 2}}, ctx) is False
+        # composites
+        assert evaluate({"all_of": [{"explored_pct_gte": 50}, {"within_ticks": 5000}]}, ctx)
+        assert evaluate({"any_of": [{"explored_pct_gte": 99}, {"units_killed_gte": 3}]}, ctx)
+        assert evaluate({"not": {"explored_pct_gte": 99}}, ctx) is True
+
+    def test_win_condition_rejects_unknown_keys(self):
+        from openra_bench.scenarios.win_conditions import WinCondition
+
+        with pytest.raises(ValueError):
+            WinCondition(definitely_not_a_predicate=1)
+        with pytest.raises(ValueError):
+            WinCondition(all_of=[{"explored_pct_gte": 1}], explored_pct_gte=2)
+
+    def test_eval_core_win_and_loss_wiring_deterministic(self):
+        """Pin the win/fail plumbing without depending on bot skill:
+        a trivially-true win condition => 'win' on turn 1; a trivially-
+        true fail condition => 'loss'."""
+        from openra_bench.eval_core import run_level
+        from openra_bench.scenarios.schema import ScenarioPack
+
+        base = {
+            "agent": {"faction": "allies"},
+            "enemy": {"faction": "soviet"},
+            "tools": ["move_units", "attack_unit"],
+            "actors": [
+                {"type": "jeep", "owner": "agent", "position": [5, 5], "count": 2},
+                {"type": "e1", "owner": "enemy", "position": [60, 20], "stance": 2},
+            ],
+            "termination": {"max_ticks": 8000},
+        }
+        meta = {
+            "id": "selftest-wiring",
+            "title": "Self Test",
+            "capability": "action",
+            "real_world_meaning": "deterministic plumbing check for the runner",
+            "robotics_analogue": "unit-test harness",
+            "author": "ci",
+        }
+
+        def pack(win, fail=None):
+            lvl = {
+                "description": "wiring check level",
+                "overrides": {},
+                "win_condition": win,
+                "max_turns": 3,
+            }
+            if fail:
+                lvl["fail_condition"] = fail
+            return ScenarioPack(
+                meta=meta, base_map="rush-hour-arena", base=base,
+                levels={"easy": lvl, "medium": lvl, "hard": lvl},
+            )
+
+        win_pack = pack({"after_ticks": 0})  # always true once ticks >= 0
+        res = run_level(win_pack.compile("easy"), bot_to_agent(idle_bot), seed=3)
+        assert res.outcome == "win"
+        assert res.signals.outcome == 1.0
+        assert res.turns == 1
+        assert len(res.trace) == res.turns
+
+        loss_pack = pack({"explored_pct_gte": 999}, fail={"after_ticks": 0})
+        res2 = run_level(loss_pack.compile("easy"), bot_to_agent(idle_bot), seed=3)
+        assert res2.outcome == "loss"
+        assert res2.signals.outcome == 0.0
+
+    @pytest.mark.parametrize(
+        "pack_file,level",
+        [
+            ("perception-frontier-reading.yaml", "easy"),
+            ("reasoning-frontier-commit.yaml", "easy"),
+            ("action-multiunit-coordination.yaml", "easy"),
+        ],
+    )
+    def test_authored_packs_run_end_to_end(self, pack_file, level):
+        from openra_bench.eval_core import run_level, scripted_explore_agent
+        from openra_bench.scenarios import load_pack
+        from openra_bench.scenarios.loader import PACKS_DIR, compile_level
+
+        pack = load_pack(PACKS_DIR / pack_file)
+        compiled = compile_level(pack, level)
+        assert compiled.map_supported, "authored packs must target a Rust-loadable map"
+        res = run_level(compiled, scripted_explore_agent, seed=1)
+        assert res.outcome in {"win", "draw", "loss"}
+        assert 1 <= res.turns <= compiled.max_turns
+        assert len(res.trace) == res.turns
+        assert res.signals.game_tick > 0
+        assert res.signals.as_reward_kwargs()["outcome"] in (0.0, 0.5, 1.0)
+
+
+def bot_to_agent(bot):
+    """Adapt an obs-only rule bot to the eval_core agent_fn signature
+    `(render_state, Command) -> [Command]`. The render_state carries
+    unit ids the bot needs under units_summary."""
+
+    def _agent(render_state, Command):
+        up = {
+            str(u["id"]): {"cell_x": u["cell_x"], "cell_y": u["cell_y"]}
+            for u in render_state.get("units_summary", [])
+        }
+        return bot({"unit_positions": up, "enemy_positions": []})
+
+    return _agent