ctan
3c3e30634e
- LLMPolicy is now Callable[[Observation], tuple[Action, ...]]: the snapshot is the policy's sole information source; the live CombatantState is no longer passed to new-style policies. - CombatEngine._llm_adapter keeps a compatibility path for pre-D1 two-argument callables: the signature is inspected ONCE at wrap time (_accepts_state) and legacy policies still receive (observation, state). - Observation.to_dict() carries schema_version (OBSERVATION_SCHEMA_VERSION=1) as its first pinned key; bump only on removal/meaning change. - Pinned keyset tests updated (test_observation, test_observation_enriched); new tests: observation-only e2e, legacy 2-arg e2e through the adapter, schema_version pin + JSON-serializability. - README: documents the new entry signature and the schema_version key.
566 lines
20 KiB
Python
566 lines
20 KiB
Python
"""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, carries the pinned ``schema_version``
|
|
key, and exposes no ``engine.*`` callables.
|
|
- The ``llm_policy`` adapter builds the Observation at the point of policy
|
|
invocation and delegates to ``llm_policy(observation)`` (D1: observation
|
|
only); a legacy ``llm_policy(observation, state)`` callable keeps working
|
|
through the compatibility adapter.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from pf1e_simulator.combat import (
|
|
OBSERVATION_SCHEMA_VERSION,
|
|
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 observation-only llm_policy (D1 signature)
|
|
When: the battle runs
|
|
Then: the policy's decisions are honored and the transcript is coherent."""
|
|
|
|
def llm_policy(obs: Observation) -> 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)
|
|
|
|
|
|
def test_legacy_two_arg_llm_policy_still_works() -> None:
|
|
"""Given: a pre-D1 llm_policy taking (observation, state)
|
|
When: the battle runs through the compatibility adapter
|
|
Then: the policy receives BOTH arguments and its decisions are honored."""
|
|
seen_states: list[CombatantState] = []
|
|
|
|
def legacy_policy(obs: Observation, state: CombatantState) -> tuple:
|
|
seen_states.append(state)
|
|
assert state.combatant.id == obs.combatants[obs.acting_combatant]["id"]
|
|
target_id = obs.combatants[obs.pairs[0]["target_index"]]["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=legacy_policy,
|
|
)
|
|
|
|
result = engine.run()
|
|
assert seen_states, "legacy policy was never invoked"
|
|
assert any("move" in line for line in result.transcript)
|
|
|
|
|
|
def test_observation_schema_version_pinned() -> None:
|
|
"""Given: a built Observation
|
|
When: to_dict() is called
|
|
Then: the top-level key set is exactly the pinned keys and
|
|
schema_version == OBSERVATION_SCHEMA_VERSION, JSON-serializable."""
|
|
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])
|
|
|
|
data = engine._build_observation(hero_state).to_dict()
|
|
assert data["schema_version"] == OBSERVATION_SCHEMA_VERSION
|
|
assert set(data) == {
|
|
"schema_version",
|
|
"round",
|
|
"acting_combatant",
|
|
"combatants",
|
|
"pairs",
|
|
"terrain",
|
|
"elevation",
|
|
"threatened",
|
|
"ranged_threatened",
|
|
"terrain_semantics",
|
|
"condition_definitions",
|
|
"initiative_order",
|
|
}
|
|
json.dumps(data) # must not raise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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,
|
|
)
|