feat(combat): expose threat grid and terrain semantics in observation

src/pf1e_simulator/combat.py

@@ -107,7 +107,7 @@ if TYPE_CHECKING:
     from typing import Literal
 
     from pf1e_simulator.grid import Grid
-    from pf1e_simulator.map import Pos
+    from pf1e_simulator.map import Pos, TerrainType
     from pf1e_simulator.models import AttackSpec, Combatant, DamageReduction
     from pf1e_simulator.rng import Rng
     from pf1e_simulator.spells import SpellSpec
@@ -381,7 +381,15 @@ class Observation:
     ``pairs`` holds one dict per (acting combatant x each other combatant) with
     ``target_index`` disambiguating the target. ``terrain`` and ``elevation``
     are height x width layers aligned on the same dimensions (``elevation`` is
-    ``()`` when the grid has no elevation layer).
+    ``()`` when the grid has no elevation layer). ``threatened`` is a height x
+    width boolean grid whose cells are True when threatened by at least one
+    ENEMY of the acting combatant (active, opposite side, with a melee
+    weapon); the union is bounded to the grid — out-of-grid threatened cells
+    are dropped. ``terrain_semantics`` is one dict per distinct terrain type
+    on the map (sorted by type for determinism), carrying
+    ``{type, move_cost, blocks_los, cover}`` — derived from
+    ``grid.terrain_at((r, c))`` over the entire map (the Grid does not expose
+    ``MapSpec.legend``).
     """
 
     round: int
@@ -390,6 +398,8 @@ class Observation:
     pairs: tuple[dict[str, object], ...]
     terrain: tuple[tuple[str | None, ...], ...]
     elevation: tuple[tuple[int, ...], ...]
+    threatened: tuple[tuple[bool, ...], ...] = ()
+    terrain_semantics: tuple[dict[str, object], ...] = ()
 
     def to_dict(self) -> dict[str, object]:
         """Return a JSON-serializable dict with the pinned top-level keys."""
@@ -400,6 +410,8 @@ class Observation:
             "pairs": self.pairs,
             "terrain": self.terrain,
             "elevation": self.elevation,
+            "threatened": self.threatened,
+            "terrain_semantics": self.terrain_semantics,
         }
 
 
@@ -587,6 +599,8 @@ class CombatEngine:
             )
         else:
             elevation = ()
+        threatened = self._threat_grid(state)
+        terrain_semantics = self._terrain_semantics()
         return Observation(
             round=self._current_round,
             acting_combatant=acting_index,
@@ -594,6 +608,56 @@ class CombatEngine:
             pairs=pairs,
             terrain=terrain,
             elevation=elevation,
+            threatened=threatened,
+            terrain_semantics=terrain_semantics,
+        )
+
+    def _threat_grid(self, state: CombatantState) -> tuple[tuple[bool, ...], ...]:
+        """Build the height x width bool grid of cells threatened by enemies.
+
+        Union of ``threatened_squares(e)`` for every enemy ``e`` of ``state``
+        (active, opposite side, with a usable melee weapon). The union is
+        bounded to the grid: cells produced by the public helper that lie
+        outside the map are dropped (helper output is in 5-ft-square coords
+        and may overflow on small maps).
+        """
+        enemies = (
+            s for s in self._states
+            if s is not state
+            and s.active
+            and s.side != state.side
+            and any(w.kind == "melee" for w in s.usable_attacks())
+        )
+        threatened: set[Pos] = set()
+        for enemy in enemies:
+            threatened |= self.threatened_squares(enemy)
+        return tuple(
+            tuple((r, c) in threatened for c in range(self._grid.width))
+            for r in range(self._grid.height)
+        )
+
+    def _terrain_semantics(self) -> tuple[dict[str, object], ...]:
+        """One dict per distinct ``TerrainType`` on the map (sorted by type).
+
+        Derived by deduplicating ``grid.terrain_at`` over the entire map: the
+        Grid stores resolved ``TerrainType`` objects in ``_cells`` but does
+        not expose ``MapSpec.legend`` directly, so iterating the cells is
+        the safest source. The Grid is guaranteed to hold at least one
+        cell (constructor enforces non-empty ``cells``).
+        """
+        seen: dict[str, TerrainType] = {}
+        for r in range(self._grid.height):
+            for c in range(self._grid.width):
+                cell = self._grid.terrain_at((r, c))
+                seen.setdefault(cell.type, cell)
+        return tuple(
+            {
+                "type": cell.type,
+                "move_cost": cell.move_cost,
+                "blocks_los": cell.blocks_los,
+                "cover": cell.cover,
+            }
+            for cell in (seen[name] for name in sorted(seen))
         )
 
     def _combatant_snapshot(self, state: CombatantState) -> dict[str, object]:

tests/test_observation_enriched.py

@@ -675,3 +675,342 @@ def test_observation_enriched_pair_existing_keys_preserved() -> None:
         "target_threatens_me",
         "enemies_threatening_me",
     }
+
+
+# ---------------------------------------------------------------------------
+# 11. E5: top-level threat grid (union of enemy threatened squares)
+# ---------------------------------------------------------------------------
+
+
+def make_reach_combatant(cid: str, *, reach_ft: int = 10) -> Combatant:
+    """Build a combatant whose melee weapon has the requested reach.
+
+    ``reach_ft=10`` is a typical reach weapon (e.g. longspear): threatens
+    the ring at 2 squares (5-10-5) and NOT the 1-square dead zone.
+    ``reach_ft=5`` is a non-reach melee weapon: threatens the disk of 1 cell.
+    """
+    attack = AttackSpec(
+        id=f"{cid}-w",
+        name="longspear" if reach_ft > 5 else "short sword",
+        kind="melee",
+        attack_bonus=5,
+        damage=[DamageComponent(formula=parse_dice("1d8"), types=["piercing"])],
+        reach_ft=reach_ft,
+    )
+    return Combatant(
+        id=cid,
+        name=cid,
+        level=1,
+        size="Medium",
+        abilities=AbilityScores(
+            str_score=10,
+            dex_score=10,
+            con_score=12,
+            int_score=10,
+            wis_score=10,
+            cha_score=10,
+        ),
+        hp_max=20,
+        ac=ACProfile(total=15, touch=15, flat_footed=15),
+        bab=5,
+        initiative_mod=0,
+        speed_land_ft=30,
+        saves=Saves(fort=0, ref=0, will=0),
+        attacks=[attack],
+    )
+
+
+def test_observation_enriched_threatened_reach_weapon_ring() -> None:
+    """Given: a reach (reach_ft=10) melee enemy at (3, 3)
+    When: an Observation is built for the actor at (3, 0)
+    Then: threatened is the 2-cell ring around (3, 3) and NOT the 1-cell
+    dead zone (ring semantics of _weapon_threatened_squares)."""
+    actor = make_combatant("actor")
+    reach_enemy = make_reach_combatant("reach_enemy", reach_ft=10)
+    actor_state = make_state(actor, "players", (3, 0))
+    reach_state = make_state(reach_enemy, "monsters", (3, 3))
+    engine = make_engine([actor_state, reach_state])
+
+    obs = _snapshot(engine, actor_state)
+    grid = obs["threatened"]
+    assert len(grid) == 8
+    assert all(len(row) == 8 for row in grid)
+
+    # Dead zone at distance 1 from (3, 3) is NOT threatened.
+    for r, c in ((2, 3), (4, 3), (3, 2), (3, 4), (2, 2), (2, 4), (4, 2), (4, 4)):
+        assert grid[r][c] is False, f"dead zone at {(r, c)} should be False"
+    # Ring at 5-10-5 distance 2 from (3, 3) IS threatened.
+    for r, c in (
+        (1, 3), (5, 3), (3, 1), (3, 5),
+        (1, 2), (1, 4), (5, 2), (5, 4),
+        (2, 1), (2, 5), (4, 1), (4, 5),
+    ):
+        assert grid[r][c] is True, f"ring at {(r, c)} should be True"
+
+
+def test_observation_enriched_threatened_non_reach_disk() -> None:
+    """Given: a non-reach melee enemy at (2, 2) on an open grid
+    When: an Observation is built for the actor
+    Then: threatened covers the 1-cell disk of distance-1 cells around the
+    enemy (no reach means no dead zone)."""
+    actor = make_combatant("actor")
+    enemy = make_combatant("enemy")
+    actor_state = make_state(actor, "players", (0, 0))
+    enemy_state = make_state(enemy, "monsters", (2, 2))
+    engine = make_engine([actor_state, enemy_state])
+
+    obs = _snapshot(engine, actor_state)
+    grid = obs["threatened"]
+    # Distance-1 disk from (2, 2): the 8 cells around it.
+    expected_true = {
+        (1, 1), (1, 2), (1, 3),
+        (2, 1),          (2, 3),
+        (3, 1), (3, 2), (3, 3),
+    }
+    for r in range(8):
+        for c in range(8):
+            assert grid[r][c] is ((r, c) in expected_true)
+
+
+def test_observation_enriched_threatened_no_melee_weapon_empty() -> None:
+    """Given: an enemy with only a ranged weapon (no melee)
+    When: an Observation is built
+    Then: threatened has no True cells (ranged threats are out of scope)."""
+    actor = make_combatant("actor")
+    archer = make_combatant("archer", ranged=True)
+    actor_state = make_state(actor, "players", (0, 0))
+    archer_state = make_state(archer, "monsters", (0, 7))
+    engine = make_engine([actor_state, archer_state])
+
+    obs = _snapshot(engine, actor_state)
+    grid = obs["threatened"]
+    assert all(cell is False for row in grid for cell in row)
+
+
+def test_observation_enriched_threatened_bounded_to_grid() -> None:
+    """Given: a melee enemy at (1, 1) on an 8x8 grid
+    When: an Observation is built
+    Then: threatened cells outside the grid do not raise — only in-bounds
+    cells appear. In particular the (0, 0) corner must be True (it's a
+    threatened distance-1 cell, in bounds) and no IndexError is raised."""
+    actor = make_combatant("actor")
+    enemy = make_combatant("enemy")
+    actor_state = make_state(actor, "players", (0, 0))
+    enemy_state = make_state(enemy, "monsters", (1, 1))
+    engine = make_engine([actor_state, enemy_state])
+
+    obs = _snapshot(engine, actor_state)
+    grid = obs["threatened"]
+    # (0, 0) is at distance 1 from (1, 1) and is threatened.
+    assert grid[0][0] is True
+    # Grid dimensions match terrain.
+    assert len(grid) == 8
+    assert all(len(row) == 8 for row in grid)
+
+
+def test_observation_enriched_threatened_ignores_inactive_enemy() -> None:
+    """Given: an enemy whose hp <= 0 (inactive)
+    When: an Observation is built
+    Then: threatened has no True cells (inactive enemies do not threaten)."""
+    actor = make_combatant("actor")
+    enemy = make_combatant("enemy")
+    actor_state = make_state(actor, "players", (0, 0))
+    enemy_state = make_state(enemy, "monsters", (0, 3))
+    enemy_state.hp = 0
+    engine = make_engine([actor_state, enemy_state])
+
+    obs = _snapshot(engine, actor_state)
+    grid = obs["threatened"]
+    assert all(cell is False for row in grid for cell in row)
+
+
+def test_observation_enriched_threatened_ignores_actor_own_threat() -> None:
+    """Given: an actor with a melee weapon and an ally (same side) with a
+    melee weapon nearby
+    When: an Observation is built
+    Then: threatened only covers ENEMY threats (ally threats excluded)."""
+    actor = make_combatant("actor")
+    ally = make_combatant("ally")
+    enemy = make_combatant("enemy")
+    actor_state = make_state(actor, "players", (0, 0))
+    ally_state = make_state(ally, "players", (0, 2))
+    enemy_state = make_state(enemy, "monsters", (0, 5))
+    engine = make_engine([actor_state, ally_state, enemy_state])
+
+    obs = _snapshot(engine, actor_state)
+    grid = obs["threatened"]
+    # Cells around the ally (0, 2) must NOT be marked (ally, not enemy).
+    for r, c in ((0, 1), (0, 3), (1, 2)):
+        assert grid[r][c] is False, f"ally-threatened {(r, c)} should be False"
+    # Cells around the enemy (0, 5) at distance 1 ARE marked.
+    for r, c in ((0, 4), (0, 6), (1, 5)):
+        assert grid[r][c] is True, f"enemy-threatened {(r, c)} should be True"
+
+
+def test_observation_enriched_threatened_union_of_two_enemies() -> None:
+    """Given: two enemies at distinct positions
+    When: an Observation is built
+    Then: threatened is the union of both disks (any True if either enemy
+    threatens it)."""
+    actor = make_combatant("actor")
+    enemy1 = make_combatant("enemy1")
+    enemy2 = make_combatant("enemy2")
+    actor_state = make_state(actor, "players", (0, 0))
+    enemy1_state = make_state(enemy1, "monsters", (2, 2))
+    enemy2_state = make_state(enemy2, "monsters", (5, 5))
+    engine = make_engine([actor_state, enemy1_state, enemy2_state])
+
+    obs = _snapshot(engine, actor_state)
+    grid = obs["threatened"]
+    # Cells in enemy1's distance-1 disk OR enemy2's distance-1 disk.
+    disk1 = {
+        (1, 1), (1, 2), (1, 3),
+        (2, 1),          (2, 3),
+        (3, 1), (3, 2), (3, 3),
+    }
+    disk2 = {
+        (4, 4), (4, 5), (4, 6),
+        (5, 4),          (5, 6),
+        (6, 4), (6, 5), (6, 6),
+    }
+    union = disk1 | disk2
+    for r in range(8):
+        for c in range(8):
+            assert grid[r][c] is ((r, c) in union)
+
+
+# ---------------------------------------------------------------------------
+# 12. E5: top-level terrain semantics (one dict per distinct terrain type)
+# ---------------------------------------------------------------------------
+
+
+def make_multi_terrain_grid() -> Grid:
+    """Build a 6x6 grid with wall, floor and pillar tiles."""
+    legend = {
+        ".": TerrainType(type="floor", move_cost=1),
+        "#": TerrainType(type="wall", move_cost=None, blocks_los=True),
+        "C": TerrainType(type="pillar", move_cost=None, cover=True),
+    }
+    spec = MapSpec(
+        name="multi",
+        terrain=(
+            "######",
+            "#....C",
+            "#..C..",
+            "#....#",
+            "C....#",
+            "######",
+        ),
+        legend=legend,
+    )
+    return Grid.from_spec(spec)
+
+
+def test_observation_enriched_terrain_semantics_all_types() -> None:
+    """Given: a grid with wall / floor / pillar tiles
+    When: an Observation is built
+    Then: terrain_semantics has one entry per distinct terrain type
+    (sorted by type for determinism), each carrying
+    type/move_cost/blocks_los/cover."""
+    hero = make_combatant("hero")
+    foe = make_combatant("foe")
+    hero_state = make_state(hero, "players", (0, 0))
+    foe_state = make_state(foe, "monsters", (3, 3))
+    grid = make_multi_terrain_grid()
+    engine = CombatEngine(ScriptedRng([]), grid, [hero_state, foe_state])
+
+    obs = _snapshot(engine, hero_state)
+    semantics = obs["terrain_semantics"]
+    assert semantics == (
+        {"type": "floor", "move_cost": 1, "blocks_los": False, "cover": False},
+        {"type": "pillar", "move_cost": None, "blocks_los": False, "cover": True},
+        {"type": "wall", "move_cost": None, "blocks_los": True, "cover": False},
+    )
+
+
+def test_observation_enriched_terrain_semantics_single_type() -> None:
+    """Given: a grid with a single terrain type (all floor)
+    When: an Observation is built
+    Then: terrain_semantics has exactly one entry, the floor."""
+    hero = make_combatant("hero")
+    foe = make_combatant("foe")
+    hero_state = make_state(hero, "players", (0, 0))
+    foe_state = make_state(foe, "monsters", (0, 3))
+    engine = make_engine([hero_state, foe_state])
+
+    obs = _snapshot(engine, hero_state)
+    assert obs["terrain_semantics"] == (
+        {"type": "floor", "move_cost": 1, "blocks_los": False, "cover": False},
+    )
+
+
+def test_observation_enriched_terrain_semantics_no_grid_iteration_needed() -> None:
+    """Given: a single-tile map containing only the wall at (0, 0)
+    When: an Observation is built
+    Then: terrain_semantics contains only the wall entry, even though no
+    passable cell is present (the function must still inspect the grid)."""
+    legend = {"#": TerrainType(type="wall", move_cost=None, blocks_los=True)}
+    spec = MapSpec(name="allwall", terrain=("##", "##"), legend=legend)
+    grid = Grid.from_spec(spec)
+    hero = make_combatant("hero")
+    foe = make_combatant("foe")
+    hero_state = make_state(hero, "players", (0, 0))
+    foe_state = make_state(foe, "monsters", (0, 0))
+    engine = CombatEngine(ScriptedRng([]), grid, [hero_state, foe_state])
+
+    obs = _snapshot(engine, hero_state)
+    assert obs["terrain_semantics"] == (
+        {"type": "wall", "move_cost": None, "blocks_los": True, "cover": False},
+    )
+
+
+def test_observation_enriched_terrain_semantics_no_callable_no_share() -> None:
+    """Given: an Observation built over a multi-terrain grid
+    When: to_dict() is called and json.dumps is invoked
+    Then: serialization succeeds and no TerrainType / callable leaks into
+    the snapshot (only builtins)."""
+    hero = make_combatant("hero")
+    foe = make_combatant("foe")
+    hero_state = make_state(hero, "players", (0, 0))
+    foe_state = make_state(foe, "monsters", (0, 0))
+    grid = make_multi_terrain_grid()
+    engine = CombatEngine(ScriptedRng([]), grid, [hero_state, foe_state])
+
+    obs = engine._build_observation(hero_state)
+    data = obs.to_dict()
+    json.dumps(data)  # must not raise
+    for entry in data["terrain_semantics"]:
+        assert set(entry) == {"type", "move_cost", "blocks_los", "cover"}
+
+
+# ---------------------------------------------------------------------------
+# 13. E5: top-level new fields don't disturb the L1/L2/L3/L4 contract
+# ---------------------------------------------------------------------------
+
+
+def test_observation_enriched_e5_existing_keys_preserved() -> None:
+    """Given: an Observation built for the L5 fields
+    When: the top-level dict is inspected
+    Then: all pre-existing top-level keys remain (round, acting_combatant,
+    combatants, pairs, terrain, elevation) and the two new keys
+    (threatened, terrain_semantics) are added — exact key set."""
+    hero = make_combatant("hero")
+    foe = make_combatant("foe")
+    hero_state = make_state(hero, "players", (0, 0))
+    foe_state = make_state(foe, "monsters", (0, 3))
+    engine = make_engine([hero_state, foe_state])
+
+    obs = _snapshot(engine, hero_state)
+    assert set(obs) == {
+        "round",
+        "acting_combatant",
+        "combatants",
+        "pairs",
+        "terrain",
+        "elevation",
+        "threatened",
+        "terrain_semantics",
+    }
+    # combatants / pairs untouched by E5
+    assert len(obs["combatants"]) == 2
+    assert len(obs["pairs"]) == 1