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.
This commit is contained in:
2026-09-21 12:07:23 +02:00
parent e5c4892a57
commit de88543c5c
7 changed files with 734 additions and 27 deletions
+125 -24
View File
@@ -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