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

feat(phase2): strategy compiler and sweep runner (P2)

Browse Source
This commit is contained in:
Thien An
2026-09-22 13:46:05 +02:00
parent 15d24be638
commit dbd1a5822e
4 changed files with 818 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_phase2_strategy.py
+231
View File
@@ -0,0 +1,231 @@
"""Phase-2 Etage A strategy compiler tests (docs/phase-2-llm-architecture.md 10 P2.1).
Deterministic, offline: no network, no LLM. Pins the ``StrategySpec`` /
``compile_strategy`` contract: the all-default spec compiles to
``default_policy`` by identity, ``select_target`` honours each priority with
deterministic tie-breaks, non-default specs drive real battles through
``run_encounter`` bit-identically under a fixed seed, and invalid specs are
rejected at construction.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
from pf1e_simulator.combat import (
Action,
CombatantState,
CombatEngine,
default_policy,
)
from pf1e_simulator.grid import Grid
from pf1e_simulator.loaders import load_combatant
from pf1e_simulator.map import load_map
from pf1e_simulator.rng import SeededRng
from pf1e_simulator.runner import EncounterSpec, Side, run_encounter
from pf1e_simulator.spells import SPELLS_DIR, load_spell_registry
from pf1e_simulator.strategy import StrategySpec, compile_strategy, select_target
if TYPE_CHECKING:
from pf1e_simulator.models import Combatant
ROOT = Path(__file__).resolve().parents[1]
MAP = ROOT / "data" / "maps" / "sample_arena.yaml"
_MAP_SPEC = load_map(MAP)
_SPELLS = load_spell_registry(SPELLS_DIR)
def _c(path: str) -> Combatant:
return load_combatant(ROOT / "data" / path)
def _spec() -> EncounterSpec:
"""A fast 2v2 (goblin archer + orc vs orc + goblin) on the sample arena."""
return EncounterSpec(
map=_MAP_SPEC,
round_cap=20,
sides=(
Side(
name="players",
combatants=(_c("monsters/goblin.json"), _c("monsters/orc.json")),
zone="A",
),
Side(
name="monsters",
combatants=(_c("monsters/orc.json"), _c("monsters/goblin.json")),
zone="B",
),
),
)
def _state(
cid: str, side: str, pos: tuple[int, int], hp: int = 10, path: str = "monsters/orc.json"
) -> CombatantState:
"""A bare runtime state over a loaded combatant (no battle run)."""
combatant = _c(path).model_copy(update={"id": cid})
return CombatantState(combatant=combatant, side=side, pos=pos, hp=hp, resources_left={})
def _mini_engine(states: list[CombatantState]) -> CombatEngine:
"""A live engine over the 2v2 map with the given states (no battle run)."""
return CombatEngine(SeededRng(1), Grid.from_spec(_MAP_SPEC), states, spell_registry=_SPELLS)
def test_compile_strategy_default_is_default_policy() -> None:
"""The all-default spec compiles to ``default_policy`` by identity."""
assert compile_strategy(StrategySpec()) is default_policy
def test_select_target_nearest() -> None:
"""nearest picks the closest active enemy (ties keep list order)."""
orc = _state("orc1", "monsters", (1, 1))
goblin = _state("goblin1", "players", (1, 2), hp=6, path="monsters/goblin.json")
far_orc = _state("orc2", "players", (1, 16), hp=6)
engine = _mini_engine([orc, goblin, far_orc])
assert select_target(engine, orc, "nearest") is goblin
def test_select_target_weakest_ac() -> None:
"""weakest_ac picks the lowest total AC (goblin 16 > orc 13)."""
orc = _state("orc1", "monsters", (1, 1))
goblin = _state("goblin1", "players", (1, 2), hp=6, path="monsters/goblin.json")
far_orc = _state("orc2", "players", (1, 16), hp=6)
engine = _mini_engine([orc, goblin, far_orc])
assert select_target(engine, orc, "weakest_ac") is far_orc
def test_select_target_lowest_hp() -> None:
"""lowest_hp picks the most wounded active enemy."""
orc = _state("orc1", "monsters", (1, 1))
healthy = _state("goblin1", "players", (1, 2), hp=6, path="monsters/goblin.json")
wounded = _state("goblin2", "players", (1, 16), hp=1, path="monsters/goblin.json")
engine = _mini_engine([orc, healthy, wounded])
assert select_target(engine, orc, "lowest_hp") is wounded
def test_select_target_no_active_enemy() -> None:
"""No active enemy -> None for every priority."""
orc = _state("orc1", "monsters", (1, 1))
dying = _state("goblin1", "players", (1, 2), hp=-5, path="monsters/goblin.json")
engine = _mini_engine([orc, dying])
assert select_target(engine, orc, "nearest") is None
assert select_target(engine, orc, "weakest_ac") is None
assert select_target(engine, orc, "lowest_hp") is None
def test_compiled_policy_melee_ladder() -> None:
"""A non-default spec still resolves the melee ladder on a live engine."""
spec = StrategySpec(name="no-charge", use_charges=False)
policy = compile_strategy(spec)
assert policy is not default_policy
orc = _state("orc1", "monsters", (1, 1))
goblin = _state("goblin1", "players", (1, 2), hp=6, path="monsters/goblin.json")
engine = _mini_engine([orc, goblin])
assert policy(engine, orc) == (Action(kind="full_attack", target_id="goblin1"),)
def test_compiled_policy_charge_disabled() -> None:
"""use_charges=False skips the charge and falls to step + full-attack."""
spec = StrategySpec(name="no-charge", use_charges=False)
policy = compile_strategy(spec)
# The celestial wolf is melee-only (Bite, reach 5 ft): at 2 cells no
# weapon_for short-circuit, so the ladder reaches the charge branch. The
# orc sits DIAGONALLY at (2, 3): straight ahead at (1, 3) the only
# attackable charge end is 1 cell away (below the 2-cell minimum charge
# move), while from (1, 1) the diagonal end (1, 3) is a legal 2-cell
# straight line AND a single 5-foot step to (1, 2) still enters reach.
wolf = _state("wolf1", "players", (1, 1), hp=13, path="monsters/celestial_wolf.json")
orc = _state("orc1", "monsters", (2, 3))
engine = _mini_engine([wolf, orc])
assert default_policy(engine, wolf) == (Action(kind="charge", target_id="orc1"),)
assert policy(engine, wolf) == (
Action(kind="5foot_step", target_id="orc1"),
Action(kind="full_attack", target_id="orc1"),
)
def test_compiled_policy_retreats_below_threshold() -> None:
"""retreat_hp_fraction withdraws below the threshold, not at/above it."""
spec = StrategySpec(name="coward", retreat_hp_fraction=0.5)
policy = compile_strategy(spec)
orc = _state("orc1", "monsters", (1, 1), hp=2) # 2 < 0.5 * 6 = 3
goblin = _state("goblin1", "players", (1, 2), hp=6, path="monsters/goblin.json")
engine = _mini_engine([orc, goblin])
assert policy(engine, orc) == (Action(kind="withdraw", target_id="goblin1"),)
orc_ok = _state("orc2", "monsters", (1, 1), hp=3) # 3 is not below 3
engine_ok = _mini_engine([orc_ok, goblin])
assert all(a.kind != "withdraw" for a in policy(engine_ok, orc_ok))
def test_compiled_policy_spell_policy_none_skips_casting() -> None:
"""spell_policy="none" never proposes a cast even for a caster with slots."""
spec = StrategySpec(name="no-spells", spell_policy="none")
policy = compile_strategy(spec)
oni = _c("pcs/oni.json")
oni_state = CombatantState(
combatant=oni,
side="players",
pos=(1, 1),
hp=oni.hp_max,
resources_left={},
spell_slots_left={1: 4, 2: 3},
)
goblin = _state("goblin1", "monsters", (1, 2), hp=6, path="monsters/goblin.json")
engine = _mini_engine([oni_state, goblin])
# With the default spell policy the same setup casts (Shield of Faith,
# slot 1): no ally is below 50% HP and Sound Burst is an area spell, so
# _find_offensive_spell (top-level DamageEffect only) skips it.
assert default_policy(engine, oni_state) == (
Action(kind="cast_spell", target_id="oni", spell_name="Shield of Faith"),
)
# With spell_policy="none" the melee ladder runs instead.
actions = policy(engine, oni_state)
assert all(a.kind != "cast_spell" for a in actions)
def test_compiled_policy_class_abilities_disabled() -> None:
"""use_class_abilities=False skips channel/hex even when they would fire."""
spec = StrategySpec(name="no-class", use_class_abilities=False)
policy = compile_strategy(spec)
oni = _c("pcs/oni.json")
oni_state = CombatantState(
combatant=oni,
side="players",
pos=(1, 1),
hp=oni.hp_max,
resources_left={"channel_energy": 6},
spell_slots_left={1: 4, 2: 3},
)
hurt = _state("hurt", "players", (1, 3), hp=1, path="monsters/goblin.json")
goblin = _state("goblin1", "monsters", (1, 2), hp=6, path="monsters/goblin.json")
engine = _mini_engine([oni_state, hurt, goblin])
# With class abilities enabled the channel fires (ally below 50% hp).
assert default_policy(engine, oni_state) == (
Action(kind="channel_energy", target_id="goblin1", channel_mode="heal"),
)
# Disabled: the channel is skipped entirely.
actions = policy(engine, oni_state)
assert all(a.kind != "channel_energy" for a in actions)
def test_compiled_policy_deterministic_runs() -> None:
"""A compiled policy drives bit-identical battles under a fixed seed."""
spec = StrategySpec(name="focus", target_priority="weakest_ac")
policy = compile_strategy(spec)
_, runs_a = run_encounter(_spec(), runs=3, seed=7, policy=policy, collect_runs=True)
_, runs_b = run_encounter(_spec(), runs=3, seed=7, policy=policy, collect_runs=True)
assert runs_a == runs_b
def test_strategy_spec_validation() -> None:
"""Invalid specs are rejected at construction."""
with pytest.raises(ValueError, match="non-empty"):
StrategySpec(name="")
with pytest.raises(ValueError, match="within"):
StrategySpec(retreat_hp_fraction=1.5)
with pytest.raises(ValueError, match="within"):
StrategySpec(retreat_hp_fraction=-0.1)
tests/test_phase2_sweep.py
+203
View File
@@ -0,0 +1,203 @@
"""Phase-2 sweep runner tests (docs/phase-2-llm-architecture.md 10 P2.3, D9/D10).
The ``SweepRunner`` executes a grid of (scenario x strategy) cells in
parallel, each cell running ``run_encounter(..., collect_runs=True)`` under
one policy and reporting the aggregate report plus the party's danger
metrics. These tests pin the contract: scenario-major grid order, bit-for-bit
determinism, thread-pool equivalence (workers=1 vs workers=4), the
``max_total_runs`` cost bound, constructor/cell validation, the equivalence
of a ``None`` policy cell with a direct ``run_encounter`` call, and the
aligned text renderer. All cells use small n (runs <= 5) so the file stays
fast and deterministic.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
from pf1e_simulator.loaders import load_combatant
from pf1e_simulator.map import load_map
from pf1e_simulator.objectives import danger_metrics
from pf1e_simulator.phase2_spike import focus_fire_policy
from pf1e_simulator.runner import EncounterSpec, Side, run_encounter
from pf1e_simulator.sweep import SweepCell, SweepRunner, format_sweep
if TYPE_CHECKING:
from pf1e_simulator.models import Combatant
ROOT = Path(__file__).resolve().parents[1]
MAP = ROOT / "data" / "maps" / "sample_arena.yaml"
RUNS = 5
SEED = 7
ROUND_CAP = 20
PARTY = "players"
def _c(path: str) -> Combatant:
return load_combatant(ROOT / "data" / path)
def _spec(goblins: int, orcs: int) -> EncounterSpec:
return EncounterSpec(
map=load_map(MAP),
round_cap=ROUND_CAP,
sides=(
Side(
name="players",
combatants=tuple(_c("monsters/goblin.json") for _ in range(goblins)),
zone="A",
),
Side(
name="monsters",
combatants=tuple(_c("monsters/orc.json") for _ in range(orcs)),
zone="B",
),
),
)
VARIANT_A = _spec(goblins=3, orcs=2)
VARIANT_B = _spec(goblins=2, orcs=3)
SCENARIOS = (("variant_a", VARIANT_A), ("variant_b", VARIANT_B))
STRATEGIES = (("default", None), ("focus_fire_monsters", focus_fire_policy("monsters")))
def _cells() -> tuple[SweepCell, ...]:
return (
SweepCell(
scenario="variant_a",
strategy="default",
spec=VARIANT_A,
policy=None,
runs=RUNS,
seed=SEED,
),
SweepCell(
scenario="variant_b",
strategy="focus_fire_monsters",
spec=VARIANT_B,
policy=focus_fire_policy("monsters"),
runs=RUNS,
seed=SEED,
),
)
def test_run_grid_2x2_scenario_major() -> None:
"""A 2x2 grid returns 4 results in scenario-major order with sane metrics."""
runner = SweepRunner(party=PARTY, workers=2)
results = runner.run_grid(SCENARIOS, STRATEGIES, runs=RUNS, seed=SEED)
assert len(results) == 4
assert [r.cell.scenario for r in results] == [
"variant_a",
"variant_a",
"variant_b",
"variant_b",
]
assert [r.cell.strategy for r in results] == [
"default",
"focus_fire_monsters",
"default",
"focus_fire_monsters",
]
for result in results:
assert result.report.runs == RUNS
assert result.danger.runs == RUNS
assert 0.0 <= result.danger.p_survives <= 1.0
assert 0.0 <= result.danger.p_tpk <= 1.0
def test_run_is_deterministic() -> None:
"""Two runs over identical cells return identical frozen results."""
runner = SweepRunner(party=PARTY, workers=2)
cells = _cells()
first = runner.run(cells)
second = runner.run(cells)
assert first == second
def test_parallel_equivalence_workers_1_vs_4() -> None:
"""workers=1 and workers=4 produce identical results (thread-safe registry)."""
cells = _cells()
serial = SweepRunner(party=PARTY, workers=1).run(cells)
parallel = SweepRunner(party=PARTY, workers=4).run(cells)
assert serial == parallel
def test_cost_bound_raises_on_budget_exceeded() -> None:
"""Cells summing past max_total_runs raise a ValueError naming runs/budget."""
runner = SweepRunner(party=PARTY, workers=2)
cells = (
SweepCell(
scenario="a", strategy="s", spec=VARIANT_A, policy=None, runs=3, seed=1
),
SweepCell(
scenario="b", strategy="s", spec=VARIANT_B, policy=None, runs=3, seed=1
),
)
with pytest.raises(ValueError, match="runs") as excinfo:
runner.run(cells, max_total_runs=5)
assert "budget" in str(excinfo.value)
def test_validation_workers_party_runs() -> None:
"""workers=0, an empty party, and a runs=0 cell all raise ValueError."""
with pytest.raises(ValueError, match="workers"):
SweepRunner(party=PARTY, workers=0)
with pytest.raises(ValueError, match="party"):
SweepRunner(party="", workers=2)
runner = SweepRunner(party=PARTY, workers=2)
bad = SweepCell(
scenario="a", strategy="s", spec=VARIANT_A, policy=None, runs=0, seed=1
)
with pytest.raises(ValueError, match="runs"):
runner.run((bad,))
def test_policy_none_matches_direct_run_encounter() -> None:
"""A None-policy cell equals a direct run_encounter on the same scenario."""
runner = SweepRunner(party=PARTY, workers=2)
cell = SweepCell(
scenario="variant_a",
strategy="default",
spec=VARIANT_A,
policy=None,
runs=RUNS,
seed=SEED,
)
(result,) = runner.run((cell,))
report, outcomes = run_encounter(VARIANT_A, RUNS, SEED, collect_runs=True)
assert result.report == report
assert result.danger == danger_metrics(outcomes, PARTY)
def test_format_sweep_columns_and_win_pct() -> None:
"""The renderer shows header tokens, one row per result, and the party win%."""
runner = SweepRunner(party=PARTY, workers=2)
results = runner.run_grid(SCENARIOS, STRATEGIES, runs=RUNS, seed=SEED)
text = format_sweep(results, PARTY)
for token in ("Scenario", "Win%", "P(TPK)"):
assert token in text
lines = text.splitlines()
assert len(lines) == 2 + len(results)
first = results[0]
expected_win = f"{first.report.wins.get(PARTY, 0) / first.report.runs * 100:.1f}"
assert lines[2].split()[3] == expected_win