diff --git a/src/pf1e_simulator/strategy.py b/src/pf1e_simulator/strategy.py new file mode 100644 index 0000000..31929d0 --- /dev/null +++ b/src/pf1e_simulator/strategy.py @@ -0,0 +1,183 @@ +"""Etage A strategy compiler (docs/phase-2-llm-architecture.md sections 2 and 10, P2.1). + +Deterministic, additive: the engine (combat.py) is never modified. A +``StrategySpec`` is a small set of frozen knobs (target priority, retreat +threshold, spell policy, charge/class-ability toggles) compiled by +``compile_strategy`` into a ``Policy`` callable. The all-default spec compiles +to the module-level ``default_policy`` object itself (identity), so the common +case is bit-identical and zero-cost. Non-default specs mirror +``default_policy``'s decision flow with the spec's values substituted. + +``StrategySpec`` feeds the SweepRunner (sweep.py, later batch) and the LLM +NL-to-spec translation (later batch): the optimizer searches over specs, the +LLM produces specs from natural language, and this module is the single +deterministic executor both rely on. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +from pf1e_simulator.combat import ( + Action, + CombatantState, + CombatEngine, + Policy, + # The decision helpers below are module-private by convention but stable: + # the engine stays unchanged, and duplicating their spell/hex/channel + # logic here would drift. + _choose_spell_action, # pyright: ignore[reportPrivateUsage] + _emit_channel_energy, # pyright: ignore[reportPrivateUsage] + _emit_hex, # pyright: ignore[reportPrivateUsage] + _spell_combat_actions, # pyright: ignore[reportPrivateUsage] + _swift_bonus_actions, # pyright: ignore[reportPrivateUsage] + default_policy, +) + +TargetPriority = Literal["nearest", "weakest_ac", "lowest_hp"] +SpellPolicy = Literal["default", "none"] + + +@dataclass(frozen=True) +class StrategySpec: + """Deterministic strategy knobs compiled to a ``Policy`` (Etage A). + + All fields default to the behaviour of ``default_policy``; a spec built + with every default compiles to ``default_policy`` itself (identity). + """ + + name: str = "default" + target_priority: TargetPriority = "nearest" + retreat_hp_fraction: float = 0.0 # 0.0 = never retreat + spell_policy: SpellPolicy = "default" + use_charges: bool = True + use_class_abilities: bool = True + + def __post_init__(self) -> None: + if not self.name: + msg = "name must be a non-empty string" + raise ValueError(msg) + if not 0.0 <= self.retreat_hp_fraction <= 1.0: + msg = "retreat_hp_fraction must be within [0, 1]" + raise ValueError(msg) + + +def select_target( + engine: CombatEngine, state: CombatantState, priority: TargetPriority +) -> CombatantState | None: + """The active enemy selected by ``priority``; None when no enemy is active. + + Deterministic: ties break on ``engine.states`` index order. + """ + if priority == "nearest": + return engine.nearest_enemy(state) + enemies = [s for s in engine.states if s.side != state.side and s.active] + if not enemies: + return None + match priority: + case "weakest_ac": + return min( + enemies, + key=lambda s: (s.combatant.ac.total, s.hp, engine.states.index(s)), + ) + case "lowest_hp": + return min(enemies, key=lambda s: (s.hp, engine.states.index(s))) + + +def _melee_actions( + engine: CombatEngine, + state: CombatantState, + target: CombatantState, + spec: StrategySpec, +) -> tuple[Action, ...]: + """The strike ladder: full-attack, charge (if enabled), step, or approach.""" + if engine.weapon_for(state, target) is not None: + return (Action(kind="full_attack", target_id=target.combatant.id),) + if spec.use_charges: + charge = engine.can_charge(state, target) + if charge is not None and charge.weapon is not None: + return (Action(kind="charge", target_id=target.combatant.id),) + if engine.can_step5_to_attack(state, target): + return ( + Action(kind="5foot_step", target_id=target.combatant.id), + Action(kind="full_attack", target_id=target.combatant.id), + ) + return ( + Action(kind="move", target_id=target.combatant.id), + Action(kind="attack", target_id=target.combatant.id), + ) + + +def _normal_actions( + engine: CombatEngine, + state: CombatantState, + target: CombatantState, + spec: StrategySpec, +) -> tuple[Action, ...]: + """The spell/melee action sequence (without swift/free class-ability add-ons).""" + if spec.spell_policy != "none" and state.combatant.spells: + spell_action = _choose_spell_action(engine, state, target) + if spell_action is not None: + return spell_action + return _melee_actions(engine, state, target, spec) + + +def _class_ability_action( + engine: CombatEngine, state: CombatantState, target: CombatantState +) -> tuple[Action, ...] | None: + """The channel/hex replacement action when a class ability fires.""" + channel_mode = _emit_channel_energy(engine, state) + if channel_mode is not None: + return ( + Action( + kind="channel_energy", + target_id=target.combatant.id, + channel_mode=channel_mode, + ), + ) + if _emit_hex(engine, state, target): + return (Action(kind="hex", target_id=target.combatant.id, hex_name="evil_eye"),) + return None + + +def _compiled_policy(spec: StrategySpec) -> Policy: + """Build the ``Policy`` that executes ``spec`` (non-default specs only).""" + + def policy(engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]: + target = select_target(engine, state, spec.target_priority) + if target is None: + return (Action(kind="wait"),) + if engine.has_condition_flag(state, "must_flee") or ( + spec.retreat_hp_fraction > 0.0 + and state.hp < spec.retreat_hp_fraction * state.combatant.hp_max + ): + return (Action(kind="withdraw", target_id=target.combatant.id),) + if spec.use_class_abilities: + class_action = _class_ability_action(engine, state, target) + if class_action is not None: + return class_action + normal = _normal_actions(engine, state, target, spec) + if not spec.use_class_abilities: + return normal + spell_combat = _spell_combat_actions(engine, state, target) + if spell_combat is not None: + return spell_combat + swift_bonus = _swift_bonus_actions(engine, state, target, normal) + return (*swift_bonus, *normal) if swift_bonus else normal + + return policy + + +_DEFAULT_SPEC = StrategySpec() + + +def compile_strategy(spec: StrategySpec) -> Policy: + """Compile a ``StrategySpec`` into a ``Policy`` callable. + + The all-default spec returns the module-level ``default_policy`` object + itself (identity), so the common case is bit-identical and zero-cost. + """ + if spec == _DEFAULT_SPEC: + return default_policy + return _compiled_policy(spec) diff --git a/src/pf1e_simulator/sweep.py b/src/pf1e_simulator/sweep.py new file mode 100644 index 0000000..4f1bc04 --- /dev/null +++ b/src/pf1e_simulator/sweep.py @@ -0,0 +1,201 @@ +"""Parallel, cost-bounded sweep runner for Phase-2 danger cartography. + +Executes a grid of (scenario x strategy) cells (docs/phase-2-llm-architecture.md +section 10, P2.3; decisions D9/D10): each cell runs ``run_encounter(..., +collect_runs=True)`` for a fixed number of seeded runs under one policy and +reports the aggregate ``EncounterReport`` plus the party's ``DangerMetrics`` +(objectives.danger_metrics). Cells are independent, so they run on a thread +pool (D10); the spell registry is loaded once and shared read-only across +threads. The runner is strategy-agnostic: policies are passed in as callables +(``None`` = ``run_encounter``'s own ``default_policy``) and no strategy module +is imported here. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Sequence + + from pf1e_simulator.combat import Policy + from pf1e_simulator.runner import EncounterReport, EncounterSpec + from pf1e_simulator.spells import SpellSpec + +from pf1e_simulator.objectives import DangerMetrics, danger_metrics +from pf1e_simulator.runner import run_encounter +from pf1e_simulator.spells import SPELLS_DIR, load_spell_registry + + +@dataclass(frozen=True) +class SweepCell: + """One (scenario, strategy) cell of a sweep: N seeded runs with one policy.""" + + scenario: str + strategy: str + spec: EncounterSpec + policy: Policy | None # None = default_policy (run_encounter's own default) + runs: int + seed: int + + +@dataclass(frozen=True) +class SweepCellResult: + """Aggregate + danger metrics for one cell.""" + + cell: SweepCell + report: EncounterReport + danger: DangerMetrics + + +class SweepRunner: + """Run a grid of (scenario, strategy) cells in parallel, bounded by cost. + + ``party`` is the side name whose danger is measured (passed to + ``danger_metrics``); ``workers`` bounds the thread pool. The spell + registry is loaded once in ``__init__`` and shared read-only across all + cells (reloading it per cell would be wasteful; a read-only dict is safe + to share between threads). + """ + + def __init__(self, *, party: str, workers: int = 4) -> None: + if not party: + msg = "party must be a non-empty side name" + raise ValueError(msg) + if workers < 1: + msg = "workers must be >= 1" + raise ValueError(msg) + self._party: str = party + self._workers: int = workers + self._registry: dict[str, SpellSpec] = load_spell_registry(SPELLS_DIR) + + def run( + self, + cells: Sequence[SweepCell], + *, + max_total_runs: int | None = None, + ) -> tuple[SweepCellResult, ...]: + """Run every cell on the thread pool and return results in input order. + + ``max_total_runs`` caps the total Monte Carlo cost (docs section 9, + D9/D14 "Budget - plafonds petits"): when the cells' runs sum exceeds + it, a ``ValueError`` is raised before anything executes. Each cell + must run at least once. Exceptions from a cell propagate loudly - a + failing cell fails the whole sweep. + """ + total_runs = sum(cell.runs for cell in cells) + if max_total_runs is not None and total_runs > max_total_runs: + msg = ( + f"sweep budget exceeded: {total_runs} runs > " + f"{max_total_runs} max_total_runs" + ) + raise ValueError(msg) + for cell in cells: + if cell.runs < 1: + msg = ( + f"cell runs must be >= 1 " + f"(scenario={cell.scenario!r}, strategy={cell.strategy!r})" + ) + raise ValueError(msg) + with ThreadPoolExecutor(max_workers=self._workers) as pool: + futures = [pool.submit(self._run_cell, cell) for cell in cells] + results = [future.result() for future in futures] + return tuple(results) + + def _run_cell(self, cell: SweepCell) -> SweepCellResult: + """One cell's work: run the encounter and measure the party's danger.""" + report, outcomes = run_encounter( + cell.spec, + cell.runs, + cell.seed, + policy=cell.policy, + spell_registry=self._registry, + collect_runs=True, + ) + danger = danger_metrics(outcomes, self._party) + return SweepCellResult(cell=cell, report=report, danger=danger) + + def run_grid( + self, + scenarios: Sequence[tuple[str, EncounterSpec]], + strategies: Sequence[tuple[str, Policy | None]], + *, + runs: int, + seed: int, + max_total_runs: int | None = None, + ) -> tuple[SweepCellResult, ...]: + """Build the scenario-major (scenario x strategy) grid and run it. + + Cells are built in SCENARIO-major order (for each scenario, for each + strategy - mirrors the danger-map table of docs section 16), every + cell sharing the same ``runs`` and ``seed``. Names come from the + tuples; a ``None`` policy stays ``None`` (default_policy). + """ + cells = [ + SweepCell( + scenario=scenario_name, + strategy=strategy_name, + spec=spec, + policy=policy, + runs=runs, + seed=seed, + ) + for scenario_name, spec in scenarios + for strategy_name, policy in strategies + ] + return self.run(cells, max_total_runs=max_total_runs) + + +def format_sweep(results: Sequence[SweepCellResult], party: str) -> str: + """Render a sweep as aligned plain-text columns (ASCII only). + + One row per result, in input order, under a header line and a separator + row of dashes. ``party`` is the side whose danger was measured: the Win% + column shows that side's win rate (``report.wins.get(party, 0) / + report.runs``) as a percentage with one decimal. Column widths are + computed from the content; columns are separated by a single space. + """ + headers = ( + "Scenario", + "Strategy", + "Runs", + "Win%", + "Draws", + "Rounds", + "P(survive)", + "P(TPK)", + "P(>=1 death)", + "HP lost", + "Worst-decile HP", + ) + rows = [ + ( + result.cell.scenario, + result.cell.strategy, + str(result.cell.runs), + f"{result.report.wins.get(party, 0) / result.report.runs * 100:.1f}", + str(result.report.draws), + f"{result.report.avg_rounds:.1f}", + f"{result.danger.p_survives:.1%}", + f"{result.danger.p_tpk:.1%}", + f"{result.danger.p_any_death:.1%}", + f"{result.danger.expected_hp_lost:.1f}", + f"{result.danger.worst_decile_party_hp:.1f}", + ) + for result in results + ] + widths = [ + max(len(header), *(len(row[index]) for row in rows)) + for index, header in enumerate(headers) + ] + + def _line(values: Sequence[str]) -> str: + return " ".join( + value.ljust(width) for value, width in zip(values, widths, strict=True) + ) + + lines = [_line(headers), _line(["-" * width for width in widths])] + lines.extend(_line(row) for row in rows) + return "\n".join(lines) diff --git a/tests/test_phase2_strategy.py b/tests/test_phase2_strategy.py new file mode 100644 index 0000000..ab57342 --- /dev/null +++ b/tests/test_phase2_strategy.py @@ -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) diff --git a/tests/test_phase2_sweep.py b/tests/test_phase2_sweep.py new file mode 100644 index 0000000..6cd1978 --- /dev/null +++ b/tests/test_phase2_sweep.py @@ -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