feat(scenario): perception-count-the-threat — exact-count state estimation (ERQA / POMDP anchor)

Adds the exact-count cell of the perception ladder (sibling to
perception-frontier-reading / perception-target-vs-fog): state
estimation under partial observability where the EXACT census matters,
not an upper bound. The agent must DISCOVER >=K hidden enemies in fog
while preserving its force and the clock. Under-scout loses the
discovery bar (K-1 of K), over-scout (mid-map sweep through the
stance:3 inner-defender wall) loses on the clock / attrition, stall
loses on the clock — only an intended just-enough-scout policy
(probe each genuine hidden-pocket band from a safe sight-line, halt
once K is tagged) clears every tooth.

- easy: K=3 hidden in one near-east cluster (visible-by-default
rehearsal of the count task; only stall fails).
- medium: K=5 with two count:3 clusters (north y=10, south y=30),
attrition cap units_lost_lte:1 and within_ticks:3600.
- hard: K=5 + two agent spawn_point groups (NW / SW round-robined
by seed) so a memorised reading order from one spawn cannot
generalise. Same predicate, same physical roster (six in
two clusters); seed-rotated reachable-subsets mean the
'natural' first cluster flips by seed. (The Rust engine
filters only agent actors by spawn_point — enemy actors
always all place — so functional K-variation rides on the
spawn-rotated path, not on enemy-actor filtering.)

Real-world anchor: ICU triage exact census, inventory before a
binding purchase commit, ERQA state-estimation, POMDP exact-count
sub-tasks, ScienceWorld inventory.

Adds 12 tests in tests/test_perception_count.py covering the binding
contract: intended wins every level on every hard seed, stall /
under-scout / over-scout each LOSE on the levels their teeth
discriminate, the deadline is reachable in max_turns (no draw
degeneracy), and hard ships >=2 spawn_point groups. Adds the pack id
to tests/test_hard_tier.py::UPGRADED. Tests pass 387 + 12 = 399
across the suite.

openra_bench/scenarios/packs/perception-count-the-threat.yaml

@@ -0,0 +1,221 @@
+# Validate: python -m openra_bench.scenarios.validate \
+#   openra_bench/scenarios/packs/perception-count-the-threat.yaml
+# See ../CONTRIBUTING.md for the full win-condition grammar.
+#
+# PERCEPTION focus, exact-count cell of the perception ladder
+# (sibling to perception-frontier-reading / perception-target-vs-fog):
+# state estimation under partial observability where the EXACT census
+# matters, not an upper bound. The agent must DISCOVER ≥K hidden
+# enemies in the fog (`enemies_discovered_gte:K`) while preserving
+# its force (`units_lost_lte:1`) and the clock. Under-scouting (stop
+# early, K-1 discovered) LOSES the discovery bar. Over-scouting (run
+# every unit through every fog patch — including a mid-map sweep that
+# brushes stance:3 inner-defender missiles) bleeds the clock (the
+# detour through every patch never finishes the count in time) and
+# busts the attrition cap. Stalling LOSES the clock. Only an intended
+# "just-enough scout" policy — visit each genuine hidden-pocket lane
+# (north + south band) until ≥K threats are tagged from a safe
+# sight-line, then halt — passes all three teeth.
+#
+# Real-world anchor (ICU triage, inventory before a binding commit,
+# ERQA state estimation, POMDP exact-count sub-tasks, ScienceWorld
+# inventory): you need the actual count, not a safe upper bound; the
+# act of counting consumes finite probe budget and excessive probing
+# is itself costly. The capability under test is calibrating WHEN to
+# stop counting — not just "see fog → drive in".
+#
+# Hard tier (binding contract): two agent spawn_point groups (NW / SW)
+# round-robined by seed (`tests/test_hard_tier.py::UPGRADED`). The
+# physical enemy roster is six hidden threats spread across BOTH
+# halves of the map; depending on the seed-chosen spawn, the natural
+# scout path encounters a different subset first, so a memorised
+# reading order from one spawn cannot generalise. The predicate K=5
+# is the binding lower bound: the agent must reliably discover 5 of 6
+# regardless of which corner it started from. (The Rust engine
+# filters AGENT actors by spawn_point but always places ALL enemy
+# actors — see openra-data/src/oramap.rs:expand_scenario_actors — so
+# K-as-physical-count cannot vary across seeds within a single
+# scenario file. Functional K-variation is achieved by spawn-rotated
+# reachable-subsets, not by enemy-actor filtering.)
+
+meta:
+  id: perception-count-the-threat
+  title: 'Count the Threat — Discover EXACTLY Enough Hidden Enemies'
+  capability: perception
+  real_world_meaning: >
+    State estimation under partial observability where over-counting
+    wastes resources and under-counting commits blind to a binding
+    decision — you need the EXACT census, not an upper bound.
+  robotics_analogue: "Warehouse inventory pass before committing to a binding purchase order: probe enough shelves to satisfy the count, then stop — every extra probe is overhead and missing one breaks the commit"
+  benchmark_anchor:
+    - "ERQA state-estimation"
+    - "POMDP exact-count sub-tasks"
+    - "ScienceWorld inventory"
+    - "ICU triage exact census"
+    - "inventory-before-commit"
+  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:
+    # Scout team — fast jeeps + a light tank so the agent can split
+    # the count work across lanes without being one-shot by a stance:2
+    # defender that the over-scout policy provokes. Agent stance:1
+    # (ReturnFire) — scouts do NOT auto-engage on sight, so a
+    # just-enough scout glides past stance:2 inner defenders without
+    # provoking; only an explicit attack_unit (over-scout) triggers
+    # the attrition tooth.
+    - {type: jeep, owner: agent, position: [6, 6], stance: 1, count: 3}
+    - {type: 1tnk, owner: agent, position: [8, 9], stance: 1, count: 2}
+
+levels:
+  easy:
+    description: >
+      Visible-by-default rehearsal of the count task: K=3 hidden
+      threats sit on the map but in this tier the scouts spawn close
+      enough to all three that a single sweep reveals every one
+      without exposing the force. The bare skill: read the brief
+      ("≥3 hidden enemies, find them all"), move into each pocket,
+      stop. Stalling at spawn never reaches K and LOSES on the clock.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 6], stance: 1, count: 3}
+        - {type: 1tnk, owner: agent, position: [8, 9], stance: 1, count: 2}
+        # K=3 hidden threats packed in a single near-east cluster
+        # (count:3 at one cell — the engine offsets adjacent copies).
+        # A committed move-to-sightline pass (e.g. scout to (40,5))
+        # reveals all three at the next snapshot via vehicle sight≈7.
+        # stance:0 (HoldFire) so the revealed enemies never engage —
+        # discovery is FREE; the only cost the agent pays is the time
+        # spent moving and the risk of an explicit `attack_unit` (the
+        # brute "attack everything" failure mode). e3 (rocket
+        # soldier) in preference to e1 because the engine reliably
+        # surfaces e3 in the visible-enemy stream — e1 cluster
+        # instances were empirically missing from `enemy_positions`
+        # even when the scout stood on the cluster cell.
+        - {type: e3, owner: enemy, position: [40, 12], stance: 0, count: 3}
+        # Persistent passive marker so the engine does not auto-end
+        # the episode the moment the agent eliminates the three
+        # threats (would short-circuit the clock-discrimination).
+        - {type: barr, owner: enemy, position: [120, 36], stance: 0}
+    win_condition:
+      all_of:
+        - {enemies_discovered_gte: 3}
+        - {within_ticks: 2200}
+    fail_condition:
+      any_of:
+        - {after_ticks: 2201}
+        - {not: {own_units_gte: 1}}
+    max_turns: 26
+
+  medium:
+    description: >
+      Now fogged with K=5 hidden threats spread across both halves of
+      the map. Under-scouting (stop after 3–4 discoveries) LOSES the
+      discovery bar. Over-scouting (push every unit through every fog
+      patch, brushing the stance:2 inner defenders) blows the
+      attrition cap (≤1 lost). Stalling LOSES the clock. The intended
+      just-enough-scout policy splits the force across the two
+      genuine hidden-pocket lanes (NE band + SE band), tags ≥5
+      threats, and halts.
+    overrides:
+      actors:
+        - {type: jeep, owner: agent, position: [6, 6], stance: 1, count: 3}
+        - {type: 1tnk, owner: agent, position: [8, 9], stance: 1, count: 2}
+        # Two genuine hidden-pocket clusters (count:3 each — the
+        # engine packs three units around the named cell, total K=6
+        # physically placed). stance:0 (HoldFire) so discovery is
+        # FREE: the scout reveals all three at the next snapshot via
+        # vehicle sight≈7 (drive to (60,5) / (60,33) — one cell off
+        # the cluster latitude — and pass the sight cone over it
+        # without engaging). The clusters sit on opposite halves
+        # (y=12 NORTH and y=28 SOUTH), so a single one-lane scout
+        # caps at K=3 — far short of K=5. Only a genuine TWO-LANE
+        # scout (split N/S) clears the discovery bar.
+        - {type: e3, owner: enemy, position: [60, 10], stance: 0, count: 3}
+        - {type: e3, owner: enemy, position: [100, 30], stance: 0, count: 3}
+        # Inner defender wall: stance:3 (AttackAnything) e3 missiles
+        # parked mid-y x=50, between the north (y≈6-10) and south
+        # (y≈30-34) scout lanes — outside jeep sight=7 from either
+        # lane (y=19/21 are 10/12 cells from y=9 and 14/12 from y=33).
+        # A just-enough scout that stays in its lane glides past
+        # without provoking; an OVER-scout that pulls every unit
+        # through every fog patch (mid-map sweep) crosses the
+        # defender wall and bleeds time AND units — the over-scout
+        # tooth bites first via clock (long detour through every
+        # patch), and on a deeper sweep also via attrition.
+        - {type: e3, owner: enemy, position: [50, 19], stance: 3}
+        - {type: e3, owner: enemy, position: [50, 21], stance: 3}
+        # Persistent passive marker for episode-end safety (see easy).
+        - {type: barr, owner: enemy, position: [120, 36], stance: 0}
+    win_condition:
+      all_of:
+        - {enemies_discovered_gte: 5}
+        - {units_lost_lte: 1}
+        - {within_ticks: 3600}
+    fail_condition:
+      any_of:
+        - {after_ticks: 3601}
+        - {not: {units_lost_lte: 1}}
+        - {not: {own_units_gte: 1}}
+    max_turns: 40
+
+  hard:
+    description: >
+      Two seed-varied spawn groups (NW / SW) round-robin the start
+      corner; the hidden-threat roster spans BOTH halves of the map
+      so a memorised reading order from one spawn cannot generalise.
+      The agent must discover ≥5 hidden threats (of six physically
+      placed) while losing ≤1 unit, on a tighter clock. Under-scout
+      LOSES the discovery bar; over-scout brushes stance:2 inner
+      defenders and busts the attrition cap; stall LOSES the clock.
+      Only a just-enough split scout that probes both bands and stops
+      once K=5 is reached clears every tooth.
+    overrides:
+      actors:
+        # spawn_point 0 — scouts staged NW.
+        - {type: jeep, owner: agent, position: [6, 6],  stance: 1, count: 3, spawn_point: 0}
+        - {type: 1tnk, owner: agent, position: [8, 9],  stance: 1, count: 2, spawn_point: 0}
+        # spawn_point 1 — scouts staged SW (threats fixed, so which
+        # band is "on the way" flips by seed; what counted as the
+        # natural first-pocket from NW becomes the longer reach from
+        # SW and vice-versa).
+        - {type: jeep, owner: agent, position: [6, 33], stance: 1, count: 3, spawn_point: 1}
+        - {type: 1tnk, owner: agent, position: [8, 30], stance: 1, count: 2, spawn_point: 1}
+        # Two genuine hidden-pocket clusters (count:3 each, total K=6
+        # placed). stance:0 — REVEALING is free (HoldFire), the only
+        # combat cost comes from the stance:2 inner defenders that an
+        # over-scout brushes. The clusters sit on opposite halves;
+        # a one-lane sweep caps at K=3 (short of K=5). Either spawn
+        # corner reaches one cluster naturally and the other on a
+        # detour — so a memorised single-spawn reading order from NW
+        # (or SW) cannot generalise.
+        - {type: e3, owner: enemy, position: [60, 10], stance: 0, count: 3}
+        - {type: e3, owner: enemy, position: [100, 30], stance: 0, count: 3}
+        # Inner defender wall (stance:3 AttackAnything) at mid-y
+        # x=50 — outside jeep sight=7 from y=6 (10-12 cells away)
+        # and from y=33 (12-14 cells away). A just-enough scout
+        # stays in its lane; an OVER-scout that runs every unit
+        # through every fog patch bleeds time AND units crossing
+        # this wall — the over-scout teeth.
+        - {type: e3, owner: enemy, position: [50, 19], stance: 3}
+        - {type: e3, owner: enemy, position: [50, 21], stance: 3}
+        # Persistent passive marker for episode-end safety.
+        - {type: barr, owner: enemy, position: [120, 20], stance: 0}
+    win_condition:
+      all_of:
+        - {enemies_discovered_gte: 5}
+        - {units_lost_lte: 1}
+        - {within_ticks: 3600}
+    fail_condition:
+      any_of:
+        - {after_ticks: 3601}
+        - {not: {units_lost_lte: 1}}
+        - {not: {own_units_gte: 1}}
+    max_turns: 40

tests/test_hard_tier.py

@@ -35,6 +35,7 @@ UPGRADED = [
     "coordination-staggered-window",
     "harass-response-preserve",
     "strict-sequence",
+    "perception-count-the-threat",
     "perception-frontier-reading",
     "perception-target-vs-fog",
     "reasoning-frontier-commit",

tests/test_perception_count.py

@@ -0,0 +1,254 @@
+"""perception-count-the-threat — exact-count perception cell.
+
+Asserts the binding contract for the pack:
+  * intended just-enough-scout WINS every level on every hard seed,
+  * stall LOSES every level (clock),
+  * under-scout (one band only, never the other) LOSES medium/hard
+    (`enemies_discovered_gte:K` unmet),
+  * over-scout (run every unit through every fog patch — mid-map
+    sweep that brushes the stance:3 inner defenders) LOSES medium/hard
+    (clock or attrition),
+  * the timeout is REACHABLE (within_ticks ≤ 93+90·(max_turns-1) — no
+    draw degeneracy), and
+  * hard ships ≥2 agent spawn_point groups (seed-driven start).
+
+These are scripted policies — no model, no network. The discrimination
+is the whole point of the cell (calibrate WHEN to stop counting), so
+the test owns it directly. Cf. perception-frontier-reading and
+perception-target-vs-fog for the sibling perception predicates.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("openra_train", reason="Rust env wheel not installed")
+
+from openra_bench.eval_core import run_level
+from openra_bench.scenarios import load_pack
+from openra_bench.scenarios.loader import compile_level
+
+PACK_PATH = (
+    Path(__file__).parent.parent
+    / "openra_bench"
+    / "scenarios"
+    / "packs"
+    / "perception-count-the-threat.yaml"
+)
+
+
+def _stall(_rs, Command):
+    return [Command.observe()]
+
+
+def _intended_easy(rs, Command):
+    """Easy: K=3 in a single near-east cluster — drive every scout to
+    the cluster's sight-line (40, 5) and reveal all three at once."""
+    units = rs.get("units_summary", []) or []
+    if not units:
+        return [Command.observe()]
+    return [
+        Command.move_units([str(u["id"])], target_x=40, target_y=5)
+        for u in units
+    ]
+
+
+def _intended_med_hard(rs, Command):
+    """Medium/hard: two clusters — north (60,10) and south (100,30).
+    Jeeps push to the north sight-line (60,5); 1tnks detour south via
+    the x=10 column (away from the mid-map defender wall) then push
+    east to the south sight-line (100,33). Both bands' K total
+    satisfies ≥5; no unit enters the inner-defender attack radius.
+    Stable role assignment by unit type so reordering between turns
+    doesn't reshuffle who scouts which band."""
+    units = rs.get("units_summary", []) or []
+    if not units:
+        return [Command.observe()]
+    cmds = []
+    for u in units:
+        uid = str(u["id"])
+        ux, uy = u["cell_x"], u["cell_y"]
+        if u.get("type") == "jeep":
+            cmds.append(
+                Command.move_units([uid], target_x=60, target_y=5)
+            )
+        else:  # 1tnk → south band
+            if uy < 30:
+                cmds.append(
+                    Command.move_units([uid], target_x=10, target_y=33)
+                )
+            else:
+                cmds.append(
+                    Command.move_units([uid], target_x=100, target_y=33)
+                )
+    return cmds
+
+
+def _under_scout(rs, Command):
+    """One-band scout: commit ENTIRELY to whichever lane the spawn
+    starts in. Caps the agent at K=3 (one cluster only) — short of
+    K=5 — so the discovery bar is the binding LOSS clause."""
+    units = rs.get("units_summary", []) or []
+    if not units:
+        return [Command.observe()]
+    avg_y = sum(u["cell_y"] for u in units) / len(units)
+    if avg_y < 20:
+        # NW spawn → only the north band
+        return [
+            Command.move_units([str(u["id"])], target_x=60, target_y=5)
+            for u in units
+        ]
+    # SW spawn → only the south band
+    return [
+        Command.move_units([str(u["id"])], target_x=100, target_y=33)
+        for u in units
+    ]
+
+
+def _over_scout(rs, Command):
+    """Run every unit through every fog patch — concretely, push the
+    whole force into the mid-map sweep at (50,20) where the stance:3
+    inner defender wall sits. The detour never reaches the genuine
+    clusters in time AND/OR the missile wall attrites units."""
+    units = rs.get("units_summary", []) or []
+    if not units:
+        return [Command.observe()]
+    return [
+        Command.move_units([str(u["id"])], target_x=50, target_y=20)
+        for u in units
+    ]
+
+
+# ── solvency: intended wins every level + every hard seed ─────────────────
+
+
+def _seeds(level: str) -> tuple[int, ...]:
+    # hard rotates the agent spawn by `seed % spawn_point_count`, so we
+    # cover both groups across seeds 1..4 (the cadence's curation set).
+    return (1, 2, 3, 4) if level == "hard" else (1,)
+
+
+def _intended(level: str):
+    return _intended_easy if level == "easy" else _intended_med_hard
+
+
+@pytest.mark.parametrize("level", ["easy", "medium", "hard"])
+def test_intended_wins_every_level_every_seed(level):
+    pack = load_pack(PACK_PATH)
+    c = compile_level(pack, level)
+    for seed in _seeds(level):
+        res = run_level(c, _intended(level), seed=seed)
+        assert res.outcome == "win", (
+            f"intended just-enough-scout must WIN {level} seed={seed}; "
+            f"got outcome={res.outcome} seen={len(res.signals.enemies_seen_ids)} "
+            f"lost={res.signals.units_lost} tick={res.signals.game_tick}"
+        )
+
+
+# ── stability: every non-win pattern is a real reachable LOSS ─────────────
+
+
+@pytest.mark.parametrize("level", ["easy", "medium", "hard"])
+def test_stall_loses_every_level(level):
+    pack = load_pack(PACK_PATH)
+    c = compile_level(pack, level)
+    for seed in _seeds(level):
+        res = run_level(c, _stall, seed=seed)
+        assert res.outcome == "loss", (
+            f"stall must LOSE {level} seed={seed} (clock — no enemies "
+            f"discovered); got outcome={res.outcome}"
+        )
+
+
+@pytest.mark.parametrize("level", ["medium", "hard"])
+def test_under_scout_loses_medium_and_hard(level):
+    """One-band scout caps at K=3 on medium/hard (predicate K=5) →
+    `enemies_discovered_gte:5` unmet → LOSS. Easy is intentionally
+    permissive (K=3 in a single cluster, no force-split required)."""
+    pack = load_pack(PACK_PATH)
+    c = compile_level(pack, level)
+    for seed in _seeds(level):
+        res = run_level(c, _under_scout, seed=seed)
+        assert res.outcome == "loss", (
+            f"under-scout must LOSE {level} seed={seed} via discovery bar; "
+            f"got outcome={res.outcome} seen={len(res.signals.enemies_seen_ids)}"
+        )
+        assert len(res.signals.enemies_seen_ids) < 5, (
+            f"under-scout {level} seed={seed} accidentally saw ≥5 enemies "
+            f"({len(res.signals.enemies_seen_ids)}) — the one-band cap "
+            "isn't holding."
+        )
+
+
+@pytest.mark.parametrize("level", ["medium", "hard"])
+def test_over_scout_loses_medium_and_hard(level):
+    """Drive every unit into the mid-map defender wall at (50,20).
+    The detour AND/OR attrition busts at least one of:
+      * `within_ticks:3600` (clock — the sweep never finishes counting),
+      * `units_lost_lte:1` (the missile wall attrites).
+    Either way the pack emits a real LOSS, never a draw."""
+    pack = load_pack(PACK_PATH)
+    c = compile_level(pack, level)
+    for seed in _seeds(level):
+        res = run_level(c, _over_scout, seed=seed)
+        assert res.outcome == "loss", (
+            f"over-scout must LOSE {level} seed={seed}; got "
+            f"outcome={res.outcome} seen={len(res.signals.enemies_seen_ids)} "
+            f"lost={res.signals.units_lost}"
+        )
+
+
+# ── tick / max_turns alignment (no draw degeneracy) ───────────────────────
+
+
+def test_timeout_is_reachable_within_max_turns():
+    """Every level's `within_ticks` / `after_ticks` deadline must be
+    reachable inside `max_turns` — else the timeout fail-clause never
+    bites and a stall draws instead of losing. The engine advances
+    ~90 ticks per decision turn; max tick ≈ 93+90·(max_turns-1)."""
+    pack = load_pack(PACK_PATH)
+    for level in ("easy", "medium", "hard"):
+        c = compile_level(pack, level)
+        max_reachable = 93 + 90 * (c.max_turns - 1)
+        # Inspect the win + fail predicate trees for explicit tick gates.
+        for node in (
+            c.win_condition.model_dump(),
+            (c.fail_condition.model_dump() if c.fail_condition else {}),
+        ):
+            _assert_ticks_reachable(node, max_reachable, level)
+
+
+def _assert_ticks_reachable(node, max_reachable, level):
+    if not isinstance(node, dict):
+        return
+    for key, value in node.items():
+        if key in ("all_of", "any_of"):
+            for sub in value or []:
+                _assert_ticks_reachable(sub, max_reachable, level)
+        elif key == "not":
+            _assert_ticks_reachable(value, max_reachable, level)
+        elif key in ("within_ticks", "after_ticks"):
+            assert int(value) <= max_reachable, (
+                f"{level}: {key}={value} exceeds reachable tick "
+                f"{max_reachable} (max_turns={(max_reachable - 93)//90 + 1}) — "
+                "deadline never bites ⇒ draw degeneracy."
+            )
+
+
+# ── hard-tier contract: ≥2 spawn_point groups ─────────────────────────────
+
+
+def test_hard_has_two_seed_driven_spawn_groups():
+    pack = load_pack(PACK_PATH)
+    c = compile_level(pack, "hard")
+    groups = {
+        a.spawn_point
+        for a in c.scenario.actors
+        if a.owner == "agent" and a.spawn_point is not None
+    }
+    assert groups == {0, 1}, (
+        f"hard must define spawn_points {{0,1}} for seed-driven "
+        f"NW/SW alternation; got {sorted(groups)}"
+    )