From e75ab580557b0b424d8ce674143f749da76af2fc Mon Sep 17 00:00:00 2001 From: ctan Date: Sun, 6 Sep 2026 11:07:30 +0200 Subject: [PATCH] feat(combat): add serializable Observation snapshot and LLMPolicy adapter --- src/pf1e_simulator/combat.py | 148 ++++++++++- tests/test_observation.py | 497 +++++++++++++++++++++++++++++++++++ 2 files changed, 644 insertions(+), 1 deletion(-) create mode 100644 tests/test_observation.py diff --git a/src/pf1e_simulator/combat.py b/src/pf1e_simulator/combat.py index aef1c20..4e0ffbe 100644 --- a/src/pf1e_simulator/combat.py +++ b/src/pf1e_simulator/combat.py @@ -344,6 +344,41 @@ class ChargeResult: reason: str | None +@dataclass(frozen=True) +class Observation: + """Serializable snapshot of combat state for a policy/LLM. + + Built by the engine at the point of policy invocation; the snapshot is the + ONLY source of information for an ``LLMPolicy`` — it never exposes + ``engine.*`` callables nor mutable references to live state. All collections + are copied (tuples/dicts) so the snapshot is frozen and JSON-serializable. + + ``combatants`` holds one dict per combatant in ``self._states`` order; + ``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). + """ + + round: int + acting_combatant: int + combatants: tuple[dict[str, object], ...] + pairs: tuple[dict[str, object], ...] + terrain: tuple[tuple[str | None, ...], ...] + elevation: tuple[tuple[int, ...], ...] + + def to_dict(self) -> dict[str, object]: + """Return a JSON-serializable dict with the pinned top-level keys.""" + return { + "round": self.round, + "acting_combatant": self.acting_combatant, + "combatants": self.combatants, + "pairs": self.pairs, + "terrain": self.terrain, + "elevation": self.elevation, + } + + @dataclass(frozen=True) class SaveResult: """Outcome of one saving throw roll.""" @@ -446,14 +481,21 @@ class CombatEngine: *, round_cap: int = 100, policy: Policy | None = None, + llm_policy: LLMPolicy | None = None, spell_registry: dict[str, SpellSpec] | None = None, reaction_policy: ReactionPolicy | None = None, ) -> None: + if policy is not None and llm_policy is not None: + msg = "policy and llm_policy are mutually exclusive" + raise ValueError(msg) self._rng = rng self._grid = grid self._states = states self._round_cap = round_cap - self._policy = policy if policy is not None else default_policy + if llm_policy is not None: + self._policy = self._llm_adapter(llm_policy) + else: + self._policy = policy if policy is not None else default_policy self._spell_registry = spell_registry or {} self._reaction_policy = reaction_policy self._in_reaction = False @@ -479,6 +521,108 @@ class CombatEngine: """The spell registry used for spell lookups.""" return self._spell_registry + def _llm_adapter(self, llm_policy: LLMPolicy) -> Policy: + """Wrap an ``LLMPolicy`` into a ``Policy``. + + At the point of invocation (``actions = self._policy(self, state)`` in + ``_take_turn``), the Observation is built for the current state and + delegated to ``llm_policy(observation, state)``. The adapter consumes + no RNG and mutates no state. + """ + + def adapter(_engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]: + observation = self._build_observation(state) + return llm_policy(observation, state) + + return adapter + + def _build_observation(self, state: CombatantState) -> Observation: + """Build a frozen, serializable snapshot of the current combat state. + + Pure read: never mutates state nor consumes RNG. ``state`` is the + acting combatant (the one whose turn it is). + """ + acting_index = self._states.index(state) + combatants = tuple(self._combatant_snapshot(s) for s in self._states) + pairs = tuple( + self._pair_snapshot(state, target) + for target in self._states + if target is not state + ) + terrain = tuple( + tuple(self._grid.terrain_at((r, c)).type for c in range(self._grid.width)) + for r in range(self._grid.height) + ) + has_elevation = getattr(self._grid, "_elevation", None) is not None + if has_elevation: + elevation = tuple( + tuple( + self._grid.elevation_at((r, c)) for c in range(self._grid.width) + ) + for r in range(self._grid.height) + ) + else: + elevation = () + return Observation( + round=self._current_round, + acting_combatant=acting_index, + combatants=combatants, + pairs=pairs, + terrain=terrain, + elevation=elevation, + ) + + def _combatant_snapshot(self, state: CombatantState) -> dict[str, object]: + """One per-combatant dict for the Observation (frozen copies).""" + conditions = tuple( + (c.condition.name, c.rounds_left) for c in state.active_conditions + ) + tuple((c.name, 0) for c in state.conditions) + resources = dict(state.resources_left) + for level, count in state.spell_slots_left.items(): + resources[f"spell_slot_{level}"] = count + return { + "id": state.combatant.id, + "side": state.side, + "pos": state.pos, + "occupied_cells": tuple(sorted(state.occupied_cells())), + "hp": state.hp, + "hp_max": state.combatant.hp_max, + "active": state.active, + "dead": state.dead, + "conditions": conditions, + "resources_left": resources, + "weapon_broken": tuple(sorted(state.weapon_broken)), + "weapon_stolen": tuple(sorted(state.weapon_stolen)), + "grappled_with": state.grappled_with, + "moved_this_turn": state.moved_this_turn, + "power_attack": state.power_attack, + } + + def _pair_snapshot( + self, state: CombatantState, target: CombatantState + ) -> dict[str, object]: + """One per-pair dict (acting combatant x target) for the Observation.""" + distance_ft = self.creature_distance(state, target) * _SQUARE_FT + weapon = self.weapon_for(state, target) + weapon_name = weapon.name if weapon is not None else None + range_penalty = ( + self.range_penalty(weapon, distance_ft) if weapon is not None else 0 + ) + charge = self.can_charge(state, target) + can_charge = charge is not None and charge.weapon is not None + return { + "target_index": self._states.index(target), + "distance_ft": distance_ft, + "weapon_for": weapon_name, + "range_penalty": range_penalty, + "can_charge": can_charge, + "can_step5_to_attack": self.can_step5_to_attack(state, target), + "has_line_of_effect": has_line_of_effect_cells( + self._grid, state.occupied_cells(), target.occupied_cells() + ), + "flanking": self._flanking_bonus(state, target) > 0, + } + def _take_turn(self, state: CombatantState) -> bool: """Execute one combatant's full turn; return True if the battle is over.""" self._decrement_durations(state) @@ -3289,6 +3433,8 @@ def _nauseated_filter(actions: tuple[Action, ...]) -> tuple[Action, ...]: Policy = Callable[[CombatEngine, CombatantState], tuple[Action, ...]] +LLMPolicy = Callable[[Observation, CombatantState], tuple[Action, ...]] + ReactionPolicy = Callable[ [CombatEngine, CombatantState, ReactionEvent], Action | None ] diff --git a/tests/test_observation.py b/tests/test_observation.py new file mode 100644 index 0000000..f2a319a --- /dev/null +++ b/tests/test_observation.py @@ -0,0 +1,497 @@ +"""Tests for L1: serializable Observation snapshot + LLMPolicy adapter. + +Covers the pinned schema: +- Per-combatant fields (id, side, pos, occupied_cells, hp/hp_max, active/dead, + conditions, resources_left + spell_slot_* keys, weapon_broken/weapon_stolen, + grappled_with, moved_this_turn, power_attack). +- Per-pair fields (target_index, distance_ft, weapon_for, range_penalty, + can_charge, can_step5_to_attack, has_line_of_effect, flanking). +- Terrain/elevation layers. +- Frozen copies: mutating live state after construction does not change the + already-built Observation. +- ``to_dict()`` is JSON-serializable and exposes no ``engine.*`` callables. +- The ``llm_policy`` adapter builds the Observation at the point of policy + invocation and delegates to ``llm_policy(observation, state)``. +""" + +from __future__ import annotations + +import json + +import pytest + +from pf1e_simulator.combat import Action, CombatantState, CombatEngine, Observation +from pf1e_simulator.conditions import SHAKEN +from pf1e_simulator.dice import parse_dice +from pf1e_simulator.effects import ActiveCondition +from pf1e_simulator.grid import Grid +from pf1e_simulator.map import MapSpec, TerrainType +from pf1e_simulator.models import ( + AbilityScores, + ACProfile, + AttackSpec, + Combatant, + DamageComponent, + Saves, +) +from pf1e_simulator.rng import ScriptedRng + + +def make_combatant( + cid: str, + *, + size: str = "Medium", + hp: int = 20, + ac: int = 15, + attack_bonus: int = 5, + speed: int = 30, + ranged: bool = False, +) -> Combatant: + """Build a single-weapon combatant (melee by default, ranged if requested).""" + if ranged: + attack = AttackSpec( + id=f"{cid}-w", + name="short bow", + kind="ranged", + attack_bonus=attack_bonus, + damage=[DamageComponent(formula=parse_dice("1d6"), types=["piercing"])], + range_increment_ft=60, + ) + else: + attack = AttackSpec( + id=f"{cid}-w", + name="short sword", + kind="melee", + attack_bonus=attack_bonus, + damage=[DamageComponent(formula=parse_dice("1d6"), types=["slashing"])], + ) + return Combatant( + id=cid, + name=cid, + level=1, + size=size, + abilities=AbilityScores( + str_score=10, + dex_score=10, + con_score=12, + int_score=10, + wis_score=10, + cha_score=10, + ), + hp_max=hp, + ac=ACProfile(total=ac, touch=ac, flat_footed=ac), + bab=attack_bonus, + initiative_mod=0, + speed_land_ft=speed, + saves=Saves(fort=0, ref=0, will=0), + attacks=[attack], + ) + + +def make_state(combatant: Combatant, side: str, pos: tuple[int, int]) -> CombatantState: + return CombatantState(combatant=combatant, side=side, pos=pos, hp=combatant.hp_max) + + +def make_grid(*, elevation: bool = False) -> Grid: + legend = {".": TerrainType(type="floor", move_cost=1)} + spec = MapSpec( + name="test", + terrain=tuple(["." * 8] * 8), + legend=legend, + elevation=tuple(["0" * 8] * 8) if elevation else None, + ) + return Grid.from_spec(spec) + + +def make_engine(states: list[CombatantState], *, grid: Grid | None = None) -> CombatEngine: + return CombatEngine(ScriptedRng([]), grid or make_grid(), states) + + +def _snapshot(engine: CombatEngine, state: CombatantState) -> dict: + return engine._build_observation(state).to_dict() + + +# --------------------------------------------------------------------------- +# 1. Basic 1v1 schema — all combatant fields present and correct +# --------------------------------------------------------------------------- + + +def test_observation_basic_schema_1v1() -> None: + """Given: a simple 1v1 with default state + When: an Observation is built for the acting combatant + Then: every per-combatant field is present and correct.""" + 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["round"] == 0 + assert obs["acting_combatant"] == 0 + assert len(obs["combatants"]) == 2 + + c = obs["combatants"][0] + assert c["id"] == "hero" + assert c["side"] == "players" + assert c["pos"] == (0, 0) + assert c["occupied_cells"] == ((0, 0),) + assert c["hp"] == 20 + assert c["hp_max"] == 20 + assert c["active"] is True + assert c["dead"] is False + assert c["conditions"] == () + assert c["resources_left"] == {} + assert c["weapon_broken"] == () + assert c["weapon_stolen"] == () + assert c["grappled_with"] is None + assert c["moved_this_turn"] is False + assert c["power_attack"] is False + + +# --------------------------------------------------------------------------- +# 2. Conditions → (name, rounds_left) tuples +# --------------------------------------------------------------------------- + + +def test_observation_conditions() -> None: + """Given: a combatant with an active condition (shaken 2 rounds) + When: an Observation is built + Then: conditions contains (name, rounds_left).""" + hero = make_combatant("hero") + foe = make_combatant("foe") + hero_state = make_state(hero, "players", (0, 0)) + hero_state.active_conditions.append( + ActiveCondition(condition=SHAKEN, rounds_left=2, source="test") + ) + foe_state = make_state(foe, "monsters", (0, 3)) + engine = make_engine([hero_state, foe_state]) + + c = _snapshot(engine, hero_state)["combatants"][0] + assert c["conditions"] == (("shaken", 2),) + + +# --------------------------------------------------------------------------- +# 3. weapon_broken / weapon_stolen listed +# --------------------------------------------------------------------------- + + +def test_observation_weapon_broken_stolen() -> None: + """Given: a combatant with a broken and a stolen weapon + When: an Observation is built + Then: weapon_broken and weapon_stolen list the ids (sorted).""" + hero = make_combatant("hero") + foe = make_combatant("foe") + hero_state = make_state(hero, "players", (0, 0)) + hero_state.weapon_broken = {"hero-w"} + hero_state.weapon_stolen = {"hero-w"} + foe_state = make_state(foe, "monsters", (0, 3)) + engine = make_engine([hero_state, foe_state]) + + c = _snapshot(engine, hero_state)["combatants"][0] + assert c["weapon_broken"] == ("hero-w",) + assert c["weapon_stolen"] == ("hero-w",) + + +# --------------------------------------------------------------------------- +# 4. grappled_with / moved_this_turn / power_attack reflected +# --------------------------------------------------------------------------- + + +def test_observation_grappled_moved_power_attack() -> None: + """Given: a combatant that is grappled, has moved, and uses power attack + When: an Observation is built + Then: the three flags are reflected.""" + hero = make_combatant("hero") + foe = make_combatant("foe") + hero_state = make_state(hero, "players", (0, 0)) + hero_state.grappled_with = "foe" + hero_state.moved_this_turn = True + hero_state.power_attack = True + foe_state = make_state(foe, "monsters", (0, 3)) + engine = make_engine([hero_state, foe_state]) + + c = _snapshot(engine, hero_state)["combatants"][0] + assert c["grappled_with"] == "foe" + assert c["moved_this_turn"] is True + assert c["power_attack"] is True + + +# --------------------------------------------------------------------------- +# 5. Pairs: distance_ft, weapon_for, range_penalty +# --------------------------------------------------------------------------- + + +def test_observation_pairs_distance_weapon_range_penalty() -> None: + """Given: a melee attacker adjacent to a target and a ranged attacker at + the second range increment + When: an Observation is built + Then: distance_ft is 5 for adjacent, weapon_for is the weapon name, + range_penalty is -2 for a ranged weapon at the second increment.""" + # Adjacent melee pair + hero = make_combatant("hero") + foe = make_combatant("foe") + hero_state = make_state(hero, "players", (0, 0)) + foe_state = make_state(foe, "monsters", (0, 1)) + engine = make_engine([hero_state, foe_state]) + + pairs = _snapshot(engine, hero_state)["pairs"] + assert len(pairs) == 1 + p = pairs[0] + assert p["target_index"] == 1 + assert p["distance_ft"] == 5 + assert p["weapon_for"] == "short sword" + assert p["range_penalty"] == 0 + + # Ranged attacker at 15 cells (75 ft) with 60 ft increment → second increment + archer = make_combatant("archer", ranged=True) + target = make_combatant("target") + archer_state = make_state(archer, "players", (0, 0)) + target_state = make_state(target, "monsters", (0, 15)) + engine2 = make_engine([archer_state, target_state]) + + p2 = _snapshot(engine2, archer_state)["pairs"][0] + assert p2["distance_ft"] == 75 + assert p2["weapon_for"] == "short bow" + assert p2["range_penalty"] == -2 + + +# --------------------------------------------------------------------------- +# 6. Pairs: can_charge and has_line_of_effect +# --------------------------------------------------------------------------- + + +def test_observation_pairs_can_charge_and_los() -> None: + """Given: a clear straight line vs a wall blocking LoS + When: an Observation is built + Then: can_charge is True on the clear line, False behind the wall, and + has_line_of_effect is False behind blocks_los.""" + # Clear straight line: 3 cells apart, no obstacles + 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]) + p = _snapshot(engine, hero_state)["pairs"][0] + assert p["can_charge"] is True + assert p["has_line_of_effect"] is True + + # Wall between them blocks LoS and charge + legend = { + ".": TerrainType(type="floor", move_cost=1), + "#": TerrainType(type="wall", move_cost=None, blocks_los=True), + } + walled = MapSpec( + name="test", + terrain=( + "........", + "........", + "..####..", + "..#..#..", + "..#..#..", + "..####..", + "........", + "........", + ), + legend=legend, + ) + grid = Grid.from_spec(walled) + hero2 = make_combatant("hero2") + foe2 = make_combatant("foe2") + hero2_state = make_state(hero2, "players", (0, 0)) + foe2_state = make_state(foe2, "monsters", (3, 3)) + engine2 = CombatEngine(ScriptedRng([]), grid, [hero2_state, foe2_state]) + p2 = _snapshot(engine2, hero2_state)["pairs"][0] + assert p2["can_charge"] is False + assert p2["has_line_of_effect"] is False + + +# --------------------------------------------------------------------------- +# 7. Pairs: flanking +# --------------------------------------------------------------------------- + + +def test_observation_pairs_flanking() -> None: + """Given: an ally on the opposite side of the target + When: an Observation is built + Then: flanking is True for the melee attacker.""" + hero = make_combatant("hero") + foe = make_combatant("foe") + ally = make_combatant("ally") + hero_state = make_state(hero, "players", (3, 3)) + foe_state = make_state(foe, "monsters", (3, 4)) + ally_state = make_state(ally, "players", (3, 5)) + engine = make_engine([hero_state, foe_state, ally_state]) + + p = _snapshot(engine, hero_state)["pairs"][0] + assert p["target_index"] == 1 + assert p["flanking"] is True + + +# --------------------------------------------------------------------------- +# 8. Terrain / elevation layers +# --------------------------------------------------------------------------- + + +def test_observation_terrain_elevation() -> None: + """Given: a map with and without an elevation layer + When: an Observation is built + Then: terrain is filled with type names; elevation has values when present + and () when absent.""" + # With elevation + legend = {".": TerrainType(type="floor", move_cost=1)} + spec = MapSpec( + name="test", + terrain=("..", ".."), + legend=legend, + elevation=("01", "00"), + ) + 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", (1, 1)) + engine = CombatEngine(ScriptedRng([]), grid, [hero_state, foe_state]) + + obs = _snapshot(engine, hero_state) + assert obs["terrain"] == (("floor", "floor"), ("floor", "floor")) + assert obs["elevation"] == ((0, 1), (0, 0)) + + # Without elevation + hero2 = make_combatant("hero2") + foe2 = make_combatant("foe2") + hero2_state = make_state(hero2, "players", (0, 0)) + foe2_state = make_state(foe2, "monsters", (1, 1)) + engine2 = make_engine([hero2_state, foe2_state]) + obs2 = _snapshot(engine2, hero2_state) + assert obs2["terrain"] == (("floor",) * 8,) * 8 + assert obs2["elevation"] == () + + +# --------------------------------------------------------------------------- +# 9. to_dict() is JSON-serializable and exposes no engine.* callables +# --------------------------------------------------------------------------- + + +def test_observation_to_dict_json_serializable_no_engine_ref() -> None: + """Given: a built Observation + When: to_dict() is called + Then: json.dumps succeeds and the dict contains only builtins (no + engine.* callables, no mutable shared references).""" + 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 = engine._build_observation(hero_state) + data = obs.to_dict() + json.dumps(data) # must not raise + + # No callable values anywhere in the serialized dict + def assert_no_callables(value: object) -> None: + assert not callable(value), f"callable exposed in snapshot: {value!r}" + if isinstance(value, dict): + for v in value.values(): + assert_no_callables(v) + elif isinstance(value, (list, tuple)): + for v in value: + assert_no_callables(v) + + assert_no_callables(data) + + +# --------------------------------------------------------------------------- +# 10. Frozen copies: mutating live state after construction +# --------------------------------------------------------------------------- + + +def test_observation_frozen_copies() -> None: + """Given: an Observation already built + When: the live state's weapon_broken, resources_left, and the grid change + Then: the already-built Observation does not change.""" + 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 = engine._build_observation(hero_state) + data_before = obs.to_dict() + + # Mutate live state after construction + hero_state.weapon_broken.add("hero-w") + hero_state.resources_left["rage"] = 3 + hero_state.spell_slots_left[1] = 2 + hero_state.hp = 5 + hero_state.pos = (5, 5) + + data_after = obs.to_dict() + assert data_after == data_before + assert data_after["combatants"][0]["hp"] == 20 + assert data_after["combatants"][0]["pos"] == (0, 0) + assert data_after["combatants"][0]["weapon_broken"] == () + assert data_after["combatants"][0]["resources_left"] == {} + + +# --------------------------------------------------------------------------- +# 11. LLMPolicy adapter e2e +# --------------------------------------------------------------------------- + + +def test_llm_policy_wrapper_e2e() -> None: + """Given: a CombatEngine with an llm_policy that reads ONLY the observation + When: the battle runs + Then: the policy's decisions are honored and the transcript is coherent.""" + + def llm_policy(obs: Observation, state: CombatantState) -> tuple: + # Read only the observation: full_attack if the target is adjacent, + # otherwise move toward it. + target_id = obs.combatants[obs.pairs[0]["target_index"]]["id"] + if obs.pairs and obs.pairs[0]["distance_ft"] <= 5: + return (Action(kind="full_attack", target_id=target_id),) + return (Action(kind="move", target_id=target_id),) + + 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 = CombatEngine( + ScriptedRng([10, 9, 8, 15, 3, 5]), + make_grid(), + [hero_state, foe_state], + round_cap=1, + llm_policy=llm_policy, + ) + + result = engine.run() + # The hero is 3 cells away → moves first, then attacks on a later turn. + assert any("move" in line for line in result.transcript) + + +# --------------------------------------------------------------------------- +# 12. policy + llm_policy → ValueError +# --------------------------------------------------------------------------- + + +def test_llm_policy_and_policy_value_error() -> None: + """Given: both policy and llm_policy provided + When: the engine is constructed + Then: ValueError is raised.""" + hero = make_combatant("hero") + foe = make_combatant("foe") + hero_state = make_state(hero, "players", (0, 0)) + foe_state = make_state(foe, "monsters", (0, 3)) + + def llm_policy(obs: Observation, state: CombatantState) -> tuple: + return () + + with pytest.raises(ValueError, match="mutually exclusive"): + CombatEngine( + ScriptedRng([]), + make_grid(), + [hero_state, foe_state], + policy=lambda _e, _s: (), + llm_policy=llm_policy, + )