From de88543c5c19e86cf5001c6b6bd6e4be87e9403b Mon Sep 17 00:00:00 2001 From: ctan Date: Mon, 21 Sep 2026 12:07:23 +0200 Subject: [PATCH] feat(phase2): danger-cartography spike - per-run outcomes, objectives, focus-fire adversary run_encounter gains an opt-in collect_runs flag returning one RunOutcome per battle (winner, rounds, per-combatant id/side/hp/hp_max/dead) next to the EncounterReport; default behavior is unchanged. objectives.py provides the pure optimizer primitives: P(survives all), P(TPK), P(>=1 death), expected hp lost, expected rounds and a worst-decile (CVaR-style) tail metric. phase2_spike.py holds the spike-level strengthened monster policy (focus fire on the lowest-AC enemy, ties to the wounded; party side stays on default_policy). tests/test_phase2_danger_map.py prints the danger-map sketch: warband and lich scenarios measured default vs focus-fire at 100 runs / seed 11, pinning the delta deterministically. --- README.md | 14 ++- src/pf1e_simulator/objectives.py | 135 ++++++++++++++++++++ src/pf1e_simulator/phase2_spike.py | 86 +++++++++++++ src/pf1e_simulator/runner.py | 149 ++++++++++++++++++---- tests/test_objectives.py | 105 ++++++++++++++++ tests/test_phase2_danger_map.py | 193 +++++++++++++++++++++++++++++ tests/test_runner.py | 79 +++++++++++- 7 files changed, 734 insertions(+), 27 deletions(-) create mode 100644 src/pf1e_simulator/objectives.py create mode 100644 src/pf1e_simulator/phase2_spike.py create mode 100644 tests/test_objectives.py create mode 100644 tests/test_phase2_danger_map.py diff --git a/README.md b/README.md index a176d8c..d578281 100644 --- a/README.md +++ b/README.md @@ -2321,7 +2321,17 @@ Limites de sorts (conformance D6-D9, documentées plutôt que bricolées) : `win_rate_band` (bande 3σ bornée à [0, 1]). - `runner.py` — `EncounterSpec`/`Side`, `build_states` (placement en zone + désambiguïsation des ids), `run_encounter` (un `SeededRng` par run) et - agrégation en `EncounterReport` avec attrition par combattant. + agrégation en `EncounterReport` avec attrition par combattant. Option + `collect_runs=True` (Phase 2) : renvoie en plus un `RunOutcome` par combat + (vainqueur, rounds, pv/état de mort par combattant) — comportement par + défaut inchangé. +- `objectives.py` — primitives d'objectif Phase 2 (pur, sans moteur) : + P(survie totale), P(TPK), P(≥1 mort), pv perdus attendus, rounds attendus + et métrique de queue (dcile pire des pv restants) sur les `RunOutcome`. +- `phase2_spike.py` — spike de cartographie du danger : politique monstre + « focus fire » (concentration sur la cible la plus faible en CA, départage + aux pv) contre laquelle la carte de danger mesure le groupe ; la face PJ + reste sur `default_policy`. - `cli.py` — front-end argparse `pf1e-sim`, rapport français, code de sortie 2 en cas d'erreur. - `loaders/` — `combatant.py` (chargeur unifié PJ + monstres avec résolution @@ -2335,7 +2345,7 @@ Limites de sorts (conformance D6-D9, documentées plutôt que bricolées) : La gate de validation complète (tests + lint + types) : ```bash -uv run pytest -q # 1981 tests +uv run pytest -q # 2000 tests uv run ruff check src tests uv run basedpyright src # mode strict ``` diff --git a/src/pf1e_simulator/objectives.py b/src/pf1e_simulator/objectives.py new file mode 100644 index 0000000..8e05478 --- /dev/null +++ b/src/pf1e_simulator/objectives.py @@ -0,0 +1,135 @@ +"""Objective primitives for the Phase-2 optimizer and danger cartography. + +Pure functions over the per-run summaries produced by +``run_encounter(..., collect_runs=True)`` (docs/phase-2-llm-architecture.md +sections 3.1 and 3.2). These are the metrics an optimizer maximizes (party +survival) and a danger map reports per scenario - no engine access, no +state, no RNG: given the same ``RunOutcome`` tuple they always return the +same numbers, which is what makes them safe as a search objective. + +Definitions (all probabilities are empirical frequencies over the runs; +``party`` is the side name being measured, e.g. ``"players"``): + +- ``p_survives`` : P(party survives all) - no party member dead. +- ``p_tpk`` : P(total party kill) - every party member dead. +- ``p_any_death`` : P(at least one party member death). +- ``expected_hp_lost``: mean total party hit points lost (hp_max - hp, + clamped to [0, hp_max] per member: healing above max never counts + negative, and overkill on a corpse is not counted as extra loss). +- ``expected_rounds`` : mean battle length. +- ``worst_decile_party_hp`` : TAIL metric - the mean party hp *remaining* + over the worst decile of runs (the 10% with the largest hp loss, at + least 1 run): a CVaR-style measure so rare disasters are not averaged + away. A dead or dying member contributes 0 hp remaining. + +A side with no members in a run (should not happen in a normal encounter) +is treated vacuously: survives all, no TPK, 0 hp lost. +""" + +from __future__ import annotations + +import math +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Sequence + + from pf1e_simulator.runner import CombatantOutcome, RunOutcome + + +@dataclass(frozen=True) +class DangerMetrics: + """The full danger profile of one party over a batch of run outcomes.""" + + runs: int + p_survives: float + p_tpk: float + p_any_death: float + expected_hp_lost: float + expected_rounds: float + worst_decile_party_hp: float + + +def _party_members(outcome: RunOutcome, party: str) -> tuple[CombatantOutcome, ...]: + return tuple(c for c in outcome.combatants if c.side == party) + + +def _hp_lost(outcome: RunOutcome, party: str) -> int: + return sum( + min(c.hp_max, max(0, c.hp_max - c.hp)) + for c in _party_members(outcome, party) + ) + + +def _hp_remaining(outcome: RunOutcome, party: str) -> int: + return sum(max(0, c.hp) for c in _party_members(outcome, party)) + + +def p_survives(outcomes: Sequence[RunOutcome], party: str) -> float: + """P(no member of `party` is dead at the end of a run).""" + if not outcomes: + return 0.0 + hits = sum(1 for o in outcomes if not any(c.dead for c in _party_members(o, party))) + return hits / len(outcomes) + + +def p_tpk(outcomes: Sequence[RunOutcome], party: str) -> float: + """P(every member of `party` is dead at the end of a run).""" + if not outcomes: + return 0.0 + hits = 0 + for o in outcomes: + members = _party_members(o, party) + if members and all(c.dead for c in members): + hits += 1 + return hits / len(outcomes) + + +def p_any_death(outcomes: Sequence[RunOutcome], party: str) -> float: + """P(at least one member of `party` is dead at the end of a run).""" + if not outcomes: + return 0.0 + hits = sum(1 for o in outcomes if any(c.dead for c in _party_members(o, party))) + return hits / len(outcomes) + + +def expected_hp_lost(outcomes: Sequence[RunOutcome], party: str) -> float: + """Mean total `party` hit points lost per run (clamped to [0, hp_max] per member).""" + if not outcomes: + return 0.0 + return sum(_hp_lost(o, party) for o in outcomes) / len(outcomes) + + +def expected_rounds(outcomes: Sequence[RunOutcome]) -> float: + """Mean number of rounds per run (draws count their full round cap).""" + if not outcomes: + return 0.0 + return sum(o.rounds for o in outcomes) / len(outcomes) + + +def worst_decile_party_hp(outcomes: Sequence[RunOutcome], party: str) -> float: + """Mean `party` hp remaining over the worst-decile runs (tail/CVaR metric). + + The worst decile is the ``max(1, ceil(10%))`` runs with the largest hp + loss; ties keep list order. Complements ``expected_hp_lost``: the mean + hides tail disasters, this number does not. + """ + if not outcomes: + return 0.0 + tail = max(1, math.ceil(len(outcomes) / 10)) + worst = sorted(outcomes, key=lambda o: _hp_lost(o, party), reverse=True)[:tail] + return sum(_hp_remaining(o, party) for o in worst) / tail + + +def danger_metrics(outcomes: Sequence[RunOutcome], party: str) -> DangerMetrics: + """Compute the whole danger profile in one call (the optimizer's objective).""" + return DangerMetrics( + runs=len(outcomes), + p_survives=p_survives(outcomes, party), + p_tpk=p_tpk(outcomes, party), + p_any_death=p_any_death(outcomes, party), + expected_hp_lost=expected_hp_lost(outcomes, party), + expected_rounds=expected_rounds(outcomes), + worst_decile_party_hp=worst_decile_party_hp(outcomes, party), + ) diff --git a/src/pf1e_simulator/phase2_spike.py b/src/pf1e_simulator/phase2_spike.py new file mode 100644 index 0000000..9359aa0 --- /dev/null +++ b/src/pf1e_simulator/phase2_spike.py @@ -0,0 +1,86 @@ +"""Phase-2 spike: a strengthened monster policy for danger cartography. + +Spike-level, not the final ``adversary.py`` (docs/phase-2-llm-architecture.md +sections 3.2 and 10, P4.1): the danger a party faces only exists against an +adversary that plays to win, and ``default_policy`` is a weak baseline - each +monster attacks its own nearest enemy, so the party never concentrates +punishment on one member. This module provides the minimal upgrade that +proves the point: FOCUS FIRE. Every monster on the designated side +concentrates on the same party member - the active enemy with the lowest AC +(ties: lowest current hp, then initiative list order) - instead of its +nearest one. + +The party side (any side other than the designated monster side) delegates +to ``default_policy`` unchanged, so one policy object drives the whole +battle and the danger-map comparison isolates exactly the monster-side +change. Spellcasting, class abilities and maneuvers stay on the default +flow: the spike measures what target selection alone is worth. +""" + +from __future__ import annotations + +from pf1e_simulator.combat import ( + Action, + CombatantState, + CombatEngine, + Policy, + default_policy, +) + + +def focus_target(engine: CombatEngine, state: CombatantState) -> CombatantState | None: + """The enemy the whole monster side concentrates on: lowest AC, then hp, then order. + + AC is the combatant's current total AC as exposed on the frozen profile + (the spike does not re-derive buff/condition modifiers); hp is current + hit points (a dying target ranks below a healthy one). Deterministic: + ties break on the engine's initiative list order. + """ + enemies = [s for s in engine.states if s.side != state.side and s.active] + if not enemies: + return None + return min(enemies, key=lambda s: (s.combatant.ac.total, s.hp, engine.states.index(s))) + + +def _strike( + engine: CombatEngine, state: CombatantState, target: CombatantState +) -> tuple[Action, ...]: + """Attack the focus target with the default melee/ranged action ladder.""" + if engine.weapon_for(state, target) is not None: + return (Action(kind="full_attack", target_id=target.combatant.id),) + 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 focus_fire_policy(monster_side: str = "monsters") -> Policy: + """A ``Policy`` where the monster side focus-fires; every other side plays default. + + Pass to ``run_encounter(policy=focus_fire_policy("monsters"))``. The + party side delegates to ``default_policy``, so the only behavioural + change is monster target concentration. + """ + + def policy(engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]: + if state.side != monster_side: + return default_policy(engine, state) + if engine.has_condition_flag(state, "must_flee"): + target = engine.nearest_enemy(state) + if target is None: + return (Action(kind="wait"),) + return (Action(kind="withdraw", target_id=target.combatant.id),) + target = focus_target(engine, state) + if target is None: + return (Action(kind="wait"),) + return _strike(engine, state, target) + + return policy diff --git a/src/pf1e_simulator/runner.py b/src/pf1e_simulator/runner.py index 2073d11..1360717 100644 --- a/src/pf1e_simulator/runner.py +++ b/src/pf1e_simulator/runner.py @@ -10,7 +10,7 @@ attrition averages. All statistical helpers live in metrics.py. from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Literal, overload if TYPE_CHECKING: from pf1e_simulator.combat import Policy @@ -24,6 +24,7 @@ from pf1e_simulator.combat import ( CombatantState, CombatantStats, CombatEngine, + CombatResult, ) from pf1e_simulator.grid import Grid from pf1e_simulator.map import Pos, zone_cells @@ -61,6 +62,39 @@ class CombatantAttrition: healing_done: float +@dataclass(frozen=True) +class CombatantOutcome: + """Final state of one combatant at the end of one battle (per-run record). + + ``dead`` is the engine's own death condition (``CombatantState.dead``: + hp at or below the combatant's death threshold - CON for living PCs, 0 + for undead/constructs and under the simple monster-death rule). A dying + PC (hp < 0 but above -CON) is NOT dead here. ``hp`` may be negative; + ``hp_max`` is carried so objectives can compute hit points lost. + """ + + id: str + side: str + hp: int + hp_max: int + dead: bool + + +@dataclass(frozen=True) +class RunOutcome: + """Minimal per-battle outcome for the danger/optimization objectives. + + Collected only when ``run_encounter`` is called with + ``collect_runs=True``; the aggregate ``EncounterReport`` stays the + default (cheap) output. ``combatants`` covers every creature that + existed at battle end, including summons. + """ + + winner: str | None + rounds: int + combatants: tuple[CombatantOutcome, ...] + + @dataclass(frozen=True) class EncounterReport: """Aggregated results of a Monte Carlo encounter. @@ -169,6 +203,50 @@ def build_states(spec: EncounterSpec) -> list[CombatantState]: return states +def _accumulate_stats(totals: dict[str, CombatantStats], result: CombatResult) -> None: + """Add one battle's per-combatant stats into the running totals, in place. + + Accumulate whatever ids the engine reports: summoned creatures act in the + battle and appear in result.stats, so seeding from the initial states + only would KeyError on them. Ids seen for the first time get a fresh + accumulator. Duplicate summons of the same creature are disambiguated by + the engine the same way build_states does (celestial_wolf, + celestial_wolf-2, ...), so each summoned creature owns its own slot and + their attrition is split, not merged. + """ + for cid, stats in result.stats.items(): + current = totals.get(cid) + if current is None: + totals[cid] = stats + continue + totals[cid] = CombatantStats( + hits=current.hits + stats.hits, + crits=current.crits + stats.crits, + damage_dealt=current.damage_dealt + stats.damage_dealt, + damage_taken=current.damage_taken + stats.damage_taken, + healing_done=current.healing_done + stats.healing_done, + ) + + +def _run_outcome(engine: CombatEngine, result: CombatResult) -> RunOutcome: + """One battle's per-run summary, read from the engine's final live states.""" + return RunOutcome( + winner=result.winner, + rounds=result.rounds, + combatants=tuple( + CombatantOutcome( + id=state.combatant.id, + side=state.side, + hp=state.hp, + hp_max=state.combatant.hp_max, + dead=state.dead, + ) + for state in engine.states + ), + ) + + +@overload def run_encounter( spec: EncounterSpec, runs: int, @@ -178,7 +256,35 @@ def run_encounter( spell_registry: dict[str, SpellSpec] | None = None, capture_first_transcript: bool = False, monsters_die_at_zero: bool = False, -) -> EncounterReport: + collect_runs: Literal[False] = False, +) -> EncounterReport: ... + + +@overload +def run_encounter( + spec: EncounterSpec, + runs: int, + seed: int, + *, + policy: Policy | None = None, + spell_registry: dict[str, SpellSpec] | None = None, + capture_first_transcript: bool = False, + monsters_die_at_zero: bool = False, + collect_runs: Literal[True], +) -> tuple[EncounterReport, tuple[RunOutcome, ...]]: ... + + +def run_encounter( + spec: EncounterSpec, + runs: int, + seed: int, + *, + policy: Policy | None = None, + spell_registry: dict[str, SpellSpec] | None = None, + capture_first_transcript: bool = False, + monsters_die_at_zero: bool = False, + collect_runs: bool = False, +) -> EncounterReport | tuple[EncounterReport, tuple[RunOutcome, ...]]: """Run `runs` battles (one SeededRng per run) and aggregate balance metrics. The spell registry is loaded once (from ``data/spells/`` unless overridden) @@ -190,6 +296,12 @@ def run_encounter( ``monsters_die_at_zero`` enables the optional simple monster-death rule (CLI ``--simple-monster-death``): monsters die at hp <= 0; PCs keep the PF1e hp <= -CON rule. Default False keeps behavior bit-identical. + + ``collect_runs`` (Phase 2, danger cartography) additionally returns a + tuple ``(report, run_outcomes)`` where ``run_outcomes`` holds one + ``RunOutcome`` per battle (winner, rounds, final per-combatant + id/side/hp/dead). Default False keeps the signature and behavior of + every existing caller unchanged (the report alone). """ if runs <= 0: msg = "runs must be positive" @@ -201,20 +313,14 @@ def run_encounter( wins = {side.name: 0 for side in spec.sides} draws = 0 total_rounds = 0 - # Accumulate whatever ids the engine reports: summoned creatures act in the - # battle and appear in result.stats, so seeding from the initial states - # only would KeyError on them. Ids seen for the first time get a fresh - # accumulator. Duplicate summons of the same creature are disambiguated by - # the engine the same way build_states does (celestial_wolf, - # celestial_wolf-2, ...), so each summoned creature owns its own slot and - # their attrition is split, not merged. totals: dict[str, CombatantStats] = {} transcript: tuple[str, ...] | None = None + outcomes: list[RunOutcome] = [] for index in range(runs): if index > 0: states = build_states(spec) rng = SeededRng(seed + index) - result = CombatEngine( + engine = CombatEngine( rng, grid, states, @@ -222,7 +328,10 @@ def run_encounter( policy=policy, spell_registry=spell_registry, monsters_die_at_zero=monsters_die_at_zero, - ).run() + ) + result = engine.run() + if collect_runs: + outcomes.append(_run_outcome(engine, result)) if capture_first_transcript and index == 0: transcript = result.transcript if result.winner is None: @@ -230,18 +339,7 @@ def run_encounter( else: wins[result.winner] = wins.get(result.winner, 0) + 1 total_rounds += result.rounds - for cid, stats in result.stats.items(): - current = totals.get(cid) - if current is None: - totals[cid] = stats - continue - totals[cid] = CombatantStats( - hits=current.hits + stats.hits, - crits=current.crits + stats.crits, - damage_dealt=current.damage_dealt + stats.damage_dealt, - damage_taken=current.damage_taken + stats.damage_taken, - healing_done=current.healing_done + stats.healing_done, - ) + _accumulate_stats(totals, result) attrition = { cid: CombatantAttrition( hits=stats.hits / runs, @@ -252,7 +350,7 @@ def run_encounter( ) for cid, stats in totals.items() } - return EncounterReport( + report = EncounterReport( runs=runs, seed=seed, wins=wins, @@ -261,3 +359,6 @@ def run_encounter( attrition=attrition, transcript=transcript, ) + if collect_runs: + return report, tuple(outcomes) + return report diff --git a/tests/test_objectives.py b/tests/test_objectives.py new file mode 100644 index 0000000..2dde0df --- /dev/null +++ b/tests/test_objectives.py @@ -0,0 +1,105 @@ +"""Unit tests for the Phase-2 objective primitives (objectives.py). + +Synthetic RunOutcome tuples with hand-computed expectations: the metrics +are pure frequencies/means over per-run summaries, so every number below is +verifiable by eye. No engine, no RNG. +""" + +from __future__ import annotations + +from pf1e_simulator.objectives import ( + danger_metrics, + expected_hp_lost, + expected_rounds, + p_any_death, + p_survives, + p_tpk, + worst_decile_party_hp, +) +from pf1e_simulator.runner import CombatantOutcome, RunOutcome + + +def _pc(cid: str, hp: int, hp_max: int = 10) -> CombatantOutcome: + dead = hp <= -12 + return CombatantOutcome(id=cid, side="players", hp=hp, hp_max=hp_max, dead=dead) + + +def _mob(cid: str, hp: int, hp_max: int = 10) -> CombatantOutcome: + dead = hp <= -12 + return CombatantOutcome(id=cid, side="monsters", hp=hp, hp_max=hp_max, dead=dead) + + +def _run(rounds: int, *combatants: CombatantOutcome) -> RunOutcome: + return RunOutcome(winner=None, rounds=rounds, combatants=combatants) + + +def test_empty_batch_returns_zero() -> None: + assert p_survives((), "players") == 0.0 + assert p_tpk((), "players") == 0.0 + assert p_any_death((), "players") == 0.0 + assert expected_hp_lost((), "players") == 0.0 + assert expected_rounds(()) == 0.0 + assert worst_decile_party_hp((), "players") == 0.0 + + +def test_survival_tpk_and_death_rates() -> None: + # run 1: both players clean (hp 10/10, 8/10) -> survives, no death + # run 2: one player dying (hp -1, NOT dead at -CON 12), one at 3 hp + # -> survives (nobody dead), no death counted + # run 3: one player dead (hp -12), one at 0 hp -> any death, not TPK + # run 4: both dead (hp -13, -12) -> TPK + runs = ( + _run(2, _pc("a", 10), _pc("b", 8), _mob("m", 5)), + _run(3, _pc("a", -1), _pc("b", 3)), + _run(4, _pc("a", -12), _pc("b", 0)), + _run(5, _pc("a", -13), _pc("b", -12)), + ) + assert p_survives(runs, "players") == 2 / 4 + assert p_any_death(runs, "players") == 2 / 4 + assert p_tpk(runs, "players") == 1 / 4 + # monsters absent from a run: vacuously survived, never TPK + assert p_survives(runs, "monsters") == 1.0 + assert p_tpk(runs, "monsters") == 0.0 + + +def test_expected_hp_lost_floors_per_member() -> None: + # run 1: a lost 2, b lost 7 (total 9). Run 2: a at -12 counts as the + # full 10 lost (clamped), b at 10 lost nothing. Mean of the two totals + # is 9.5. + runs = ( + _run(1, _pc("a", 8), _pc("b", 3)), + _run(2, _pc("a", -12), _pc("b", 10)), + ) + assert expected_hp_lost(runs, "players") == 9.5 + assert expected_rounds(runs) == 1.5 + + +def test_worst_decile_takes_the_tail_not_the_mean() -> None: + # 20 runs: 19 clean (hp lost 0) + 1 disaster (both dead, 20 lost). + # Mean hp lost = 1.0 but the worst decile (2 runs: disaster + one clean) + # averages 10 hp remaining - the tail is visible. + clean = _run(1, _pc("a", 10), _pc("b", 10)) + disaster = _run(9, _pc("a", -12), _pc("b", -13)) + runs = (disaster, *([clean] * 19)) + assert expected_hp_lost(runs, "players") == 1.0 + assert worst_decile_party_hp(runs, "players") == (0 + 20) / 2 + + +def test_worst_decile_minimum_one_run() -> None: + runs = (_run(1, _pc("a", 4)),) + assert worst_decile_party_hp(runs, "players") == 4 + + +def test_danger_metrics_aggregates_every_primitive() -> None: + runs = ( + _run(2, _pc("a", 10), _mob("m", 0)), + _run(4, _pc("a", -12), _mob("m", -13)), + ) + m = danger_metrics(runs, "players") + assert m.runs == 2 + assert m.p_survives == 0.5 + assert m.p_tpk == 0.5 + assert m.p_any_death == 0.5 + assert m.expected_hp_lost == 5.0 + assert m.expected_rounds == 3.0 + assert m.worst_decile_party_hp == 0.0 diff --git a/tests/test_phase2_danger_map.py b/tests/test_phase2_danger_map.py new file mode 100644 index 0000000..3c7891c --- /dev/null +++ b/tests/test_phase2_danger_map.py @@ -0,0 +1,193 @@ +"""Phase-2 danger-cartography spike (docs/phase-2-llm-architecture.md 3.1/3.2, 10 P3/P4). + +Proof of value WITHOUT any LLM: the same party measured against the default +monster policy and against the strengthened focus-fire policy (spike module +pf1e_simulator.phase2_spike). A danger only exists against an adversary that +plays to win, so the DELTA between the two columns is the danger the default +Monte Carlo report hides. Prints a compact per-scenario block (party win +rate +/- 3 sigma, P(TPK), P(>=1 death), expected hp lost, rounds, worst +decile hp) and pins loose assertions: everything is deterministic under a +fixed seed, so the pinned inequalities are exact reproductions, not noise. + +Scenarios (sample arena, 100 runs, fixed seed): + warband - 4 level-4 PCs vs a thrune agent + 2 orcs + a goblin: the party + "wins" either way, but focus fire nearly doubles P(>=1 death). + lich - 2 level-2/4 PCs vs the lich + an orc: the party loses both + ways, but against the default policy nobody ever actually + DIES (downed at hp<0, above -CON); focus fire converts defeat + into real deaths and a nonzero TPK rate. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import TYPE_CHECKING + +from pf1e_simulator.combat import 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.objectives import DangerMetrics, danger_metrics +from pf1e_simulator.phase2_spike import focus_fire_policy +from pf1e_simulator.rng import SeededRng +from pf1e_simulator.runner import EncounterSpec, RunOutcome, Side, run_encounter + +if TYPE_CHECKING: + import pytest + + from pf1e_simulator.models import Combatant + +ROOT = Path(__file__).resolve().parents[1] +MAP = ROOT / "data" / "maps" / "sample_arena.yaml" + +RUNS = 100 +SEED = 11 +ROUND_CAP = 20 + + +def _c(path: str) -> Combatant: + return load_combatant(ROOT / "data" / path) + + +def _spec(pcs: list[str], monsters: list[str]) -> EncounterSpec: + return EncounterSpec( + map=load_map(MAP), + round_cap=ROUND_CAP, + sides=( + Side(name="players", combatants=tuple(_c(p) for p in pcs), zone="A"), + Side(name="monsters", combatants=tuple(_c(p) for p in monsters), zone="B"), + ), + ) + + +WARBAND = _spec( + ["pcs/jeanne.json", "pcs/tammara.json", "pcs/harvie.json", "pcs/ierlieth.json"], + ["monsters/thrune_agent.json", "monsters/orc.json", "monsters/orc.json", + "monsters/goblin.json"], +) +LICH = _spec( + ["pcs/harvie.json", "pcs/ierlieth.json"], + ["monsters/lich.json", "monsters/orc.json"], +) + + +def _emit(line: str) -> None: + sys.stdout.write(line + "\n") + + +def _mini_engine(states: list[CombatantState]) -> CombatEngine: + return CombatEngine(SeededRng(1), Grid.from_spec(WARBAND.map), states) + + +def _state(cid: str, side: str, pos: tuple[int, int], hp: int = 10) -> CombatantState: + c = _c("monsters/orc.json").model_copy(update={"id": cid}) + return CombatantState(combatant=c, side=side, pos=pos, hp=hp, resources_left={}) + + +def test_focus_fire_concentrates_on_lowest_ac_not_nearest() -> None: + """The monster ignores the adjacent tough target and converges on the squishy one.""" + policy = focus_fire_policy("monsters") + orc = _state("orc1", "monsters", (1, 1)) + tough = _c("pcs/jeanne.json") # AC 23, adjacent to the orc + squishy = _c("pcs/nairda.json") # AC 13, far away + tough_state = CombatantState( + combatant=tough, side="players", pos=(1, 2), hp=tough.hp_max, resources_left={} + ) + squishy_state = CombatantState( + combatant=squishy, side="players", pos=(1, 16), hp=squishy.hp_max, resources_left={} + ) + engine = _mini_engine([orc, tough_state, squishy_state]) + actions = policy(engine, orc) + assert all(a.target_id == "nairda" for a in actions), actions + + +def test_focus_fire_tie_break_prefers_wounded() -> None: + """Equal AC: the monster focuses the member with the lower current hp.""" + policy = focus_fire_policy("monsters") + orc = _state("orc1", "monsters", (1, 1)) + fresh = _c("monsters/goblin.json").model_copy(update={"id": "fresh"}) + hurt = _c("monsters/goblin.json").model_copy(update={"id": "hurt"}) + fresh_state = CombatantState( + combatant=fresh, side="players", pos=(5, 5), hp=fresh.hp_max, resources_left={} + ) + hurt_state = CombatantState( + combatant=hurt, side="players", pos=(5, 10), hp=1, resources_left={} + ) + engine = _mini_engine([orc, fresh_state, hurt_state]) + actions = policy(engine, orc) + assert all(a.target_id == "hurt" for a in actions), actions + + +def test_focus_fire_leaves_the_party_side_on_default_policy() -> None: + """Only the monster side changes: a party-side turn gets default_policy verbatim.""" + policy = focus_fire_policy("monsters") + pc = _state("p", "players", (1, 1)) + mob = _state("m", "monsters", (1, 8)) + engine = _mini_engine([pc, mob]) + assert policy(engine, pc) == default_policy(engine, pc) + + +def _report(name: str, m: DangerMetrics, win: float, sigma: float) -> None: + _emit( + f" {name:<9} win={win:6.1%} +/-{3 * sigma:4.1%} surv={m.p_survives:5.1%} " + f"tpk={m.p_tpk:5.1%} >=1-death={m.p_any_death:5.1%} " + f"hp-lost={m.expected_hp_lost:6.1f} rounds={m.expected_rounds:4.1f} " + f"tail-hp={m.worst_decile_party_hp:5.1f}" + ) + + +def _measure( + spec: EncounterSpec, + *, + strengthened: bool +) -> tuple[float, float, DangerMetrics, tuple[RunOutcome, ...]]: + policy = focus_fire_policy("monsters") if strengthened else None + report, outcomes = run_encounter( + spec, runs=RUNS, seed=SEED, policy=policy, collect_runs=True + ) + win = report.win_rate("players") + return win, report.win_rate_sigma("players"), danger_metrics(outcomes, "players"), outcomes + + +def test_danger_map_default_vs_strengthened(capsys: pytest.CaptureFixture[str]) -> None: + """Print the danger map sketch and pin the delta (deterministic, fixed seed).""" + _emit("=== PHASE2 SUMMARY (danger cartography spike) ===") + deltas: dict[str, tuple[float, float]] = {} + metrics: dict[str, tuple[DangerMetrics, DangerMetrics]] = {} + for name, spec in (("warband", WARBAND), ("lich", LICH)): + _emit(f"scenario {name}: {len(spec.sides[0].combatants)} PCs vs " + f"{len(spec.sides[1].combatants)} monsters, runs={RUNS}, seed={SEED}") + w0, s0, m0, _ = _measure(spec, strengthened=False) + w1, s1, m1, _ = _measure(spec, strengthened=True) + _report("default", m0, w0, s0) + _report("focusfire", m1, w1, s1) + _emit( + f" delta death-rate {m1.p_any_death - m0.p_any_death:+.1%} " + f"hp-lost {m1.expected_hp_lost - m0.expected_hp_lost:+.1f}" + ) + deltas[name] = (m0.p_any_death, m1.p_any_death) + metrics[name] = (m0, m1) + + out = capsys.readouterr().out + assert "PHASE2 SUMMARY" in out + + # Loose, deterministic pins: the strengthened adversary never hides danger. + for m0, m1 in metrics.values(): + assert 0.0 <= m1.p_any_death <= 1.0 + assert m1.p_tpk >= m0.p_tpk - 0.05 + assert m1.p_survives <= m0.p_survives + 0.05 + # warband: focus fire must strictly raise the death rate + d0, d1 = deltas["warband"] + assert d1 > d0 + # lich: default policy shows ZERO real deaths; focus fire shows some + assert metrics["lich"][0].p_any_death == 0.0 + assert metrics["lich"][1].p_any_death > 0.0 + assert metrics["lich"][1].p_tpk > 0.0 + + +def test_danger_map_is_reproducible() -> None: + """Same seed, same policy -> byte-identical per-run outcome tuples.""" + a = _measure(LICH, strengthened=True)[3] + b = _measure(LICH, strengthened=True)[3] + assert a == b diff --git a/tests/test_runner.py b/tests/test_runner.py index 70d38d3..5a2a158 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -22,7 +22,13 @@ from pf1e_simulator.models import ( DamageComponent, Saves, ) -from pf1e_simulator.runner import EncounterSpec, Side, build_states, run_encounter +from pf1e_simulator.runner import ( + EncounterReport, + EncounterSpec, + Side, + build_states, + run_encounter, +) def make_combatant( @@ -391,3 +397,74 @@ def test_run_encounter_reports_healing_done() -> None: report = run_encounter(spec, runs=5, seed=11, policy=_self_healing_policy) assert report.attrition["healer"].healing_done > 0.0 assert report.attrition["healer"].damage_taken > 0.0 + + +def _duel_spec() -> EncounterSpec: + return EncounterSpec( + map=make_map(), + sides=( + Side(name="players", combatants=(make_combatant("p1", hp=20),), zone="A"), + Side(name="monsters", combatants=(make_combatant("m1", hp=20),), zone="B"), + ), + round_cap=20, + ) + + +def test_run_encounter_default_unchanged_without_collect_runs() -> None: + """Without collect_runs the call still returns the bare EncounterReport.""" + result = run_encounter(_duel_spec(), runs=5, seed=3) + assert isinstance(result, EncounterReport) + + +def test_run_encounter_collect_runs_returns_report_and_summaries() -> None: + """collect_runs=True returns (report, per-run summaries) with consistent aggregates.""" + spec = _duel_spec() + report, outcomes = run_encounter(spec, runs=10, seed=3, collect_runs=True) + assert isinstance(report, EncounterReport) + assert len(outcomes) == 10 + wins = sum(1 for o in outcomes if o.winner == "players") + draws = sum(1 for o in outcomes if o.winner is None) + assert wins == report.wins["players"] + assert draws == report.draws + assert sum(o.rounds for o in outcomes) / 10 == report.avg_rounds + for o in outcomes: + assert {c.id for c in o.combatants} == {"p1", "m1"} + by_id = {c.id: c for c in o.combatants} + assert by_id["p1"].side == "players" + assert by_id["m1"].side == "monsters" + assert by_id["p1"].hp_max == 20 + # dead is the engine death condition: hp <= -CON (con 12 here) + for c in o.combatants: + assert c.dead == (c.hp <= -12) + + +def test_run_encounter_collect_runs_deterministic() -> None: + """Same seed -> identical per-run summaries.""" + spec = _duel_spec() + _, a = run_encounter(spec, runs=8, seed=9, collect_runs=True) + _, b = run_encounter(spec, runs=8, seed=9, collect_runs=True) + assert a == b + + +def test_run_encounter_collect_runs_dead_matches_winner() -> None: + """A side that never wins still ends runs with its member dead in the summaries.""" + spec = EncounterSpec( + map=make_map(), + sides=( + Side(name="players", combatants=(make_combatant("weak", hp=4),), zone="A"), + Side( + name="monsters", + combatants=(make_combatant("strong", hp=40, attack_bonus=10, damage="2d6"),) * 2, + zone="B", + ), + ), + round_cap=20, + ) + report, outcomes = run_encounter(spec, runs=5, seed=5, collect_runs=True) + assert report.wins["monsters"] == 5 + for o in outcomes: + weak = next(c for c in o.combatants if c.id == "weak") + # PF1e: the battle ends when weak is INACTIVE (hp < 0), not necessarily + # dead - dead must mirror the engine condition hp <= -CON, not the loss. + assert weak.hp < 0 + assert weak.dead == (weak.hp <= -12)