ctan/pf1e-simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(combat): add serializable Observation snapshot and LLMPolicy adapter

Browse Source
This commit is contained in:
Thien An
2026-09-06 11:07:30 +02:00
parent 7843756de2
commit e75ab58055
2 changed files with 644 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pf1e_simulator/combat.py
+147 -1
View File
@@ -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
]
tests/test_observation.py
+497
View File
@@ -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,
)