diff --git a/src/pf1e_simulator/combat.py b/src/pf1e_simulator/combat.py index 16e044f..051f20e 100644 --- a/src/pf1e_simulator/combat.py +++ b/src/pf1e_simulator/combat.py @@ -9,9 +9,13 @@ Phase 0 documented deviations from PF1e (conventions): straight-line charge path exists, `(5foot_step, full_attack)` when a single step brings the target into range, or `(move, attack)` otherwise (move at full speed along the Dijkstra path, then single attack). Immediate actions - (off-turn, consume next swift) are deferred; flanking, full attack, charge, - withdraw, and 5-foot step are modeled; attacks of opportunity are resolved - inline (see below). + (off-turn, consume next swift) are modeled via an opt-in ``reaction_policy`` + hook: each resolved attack, movement step, spell cast, and damage application + emits a ``ReactionEvent``; the first combatant whose policy returns an + ``Action(kind="immediate")`` executes it immediately, and its next turn's + swift action is consumed. Flanking, full attack, charge, withdraw, and + 5-foot step are modeled; attacks of opportunity are resolved inline (see + below). - A weapon with `count` > 1 ("2x Talons") resolves `count` independent attacks in one attack action; remaining swings are lost once the target drops (down or dead) mid-routine. BAB iterative attacks (full-round action @@ -246,6 +250,7 @@ class CombatantState: power_attack: bool = False grappled_with: str | None = None deed_up_close_and_deadly: bool = False + immediate_used: bool = False summon_duration: int = 0 summon_caster_id: str | None = None # Runtime weapon durability (not in the frozen data model): accumulated @@ -363,6 +368,26 @@ class Action: channel_mode: Literal["heal", "harm"] | None = None # channel energy polarity for this use +@dataclass(frozen=True) +class ReactionEvent: + """An off-turn event a ``reaction_policy`` may react to. + + ``kind`` describes what just happened: ``attack_resolved`` (a swing was + resolved, ``actor`` is the attacker, ``target`` the defender, ``damage`` + the damage dealt), ``movement_step`` (``actor`` moved one cell), + ``spell_cast`` (``actor`` finished casting at ``target``), or + ``damage_taken`` (``target`` lost ``damage`` hp from a non-attack source + such as a spell or channel energy). ``action`` is reserved for future + use (the triggering action is not threaded through the engine today). + """ + + kind: Literal["attack_resolved", "movement_step", "spell_cast", "damage_taken"] + actor: CombatantState + target: CombatantState | None = None + damage: int = 0 + action: Action | None = None + + @dataclass(frozen=True) class CombatResult: """Outcome of a full battle: winner, rounds, log, and per-combatant stats.""" @@ -403,6 +428,7 @@ class CombatEngine: round_cap: int = 100, policy: Policy | None = None, spell_registry: dict[str, SpellSpec] | None = None, + reaction_policy: ReactionPolicy | None = None, ) -> None: self._rng = rng self._grid = grid @@ -410,6 +436,8 @@ class CombatEngine: self._round_cap = round_cap self._policy = policy if policy is not None else default_policy self._spell_registry = spell_registry or {} + self._reaction_policy = reaction_policy + self._in_reaction = False self._transcript: list[str] = [] self._stats: dict[str, _LiveStats] = {s.combatant.id: _LiveStats() for s in states} self._current_round = 0 @@ -443,6 +471,9 @@ class CombatEngine: state.moved_this_turn = False state.power_attack = False actions = self._policy(self, state) + if state.immediate_used: + actions = _without_swift_bonus(actions) + state.immediate_used = False if self._has_condition_named(state, "staggered"): actions = _staggered_filter(actions) elif self.has_condition_flag(state, "cant_attack"): @@ -1407,6 +1438,11 @@ class CombatEngine: break attacker.pos = step actual_path.append(step) + self._check_reactions( + ReactionEvent(kind="movement_step", actor=attacker) + ) + if not attacker.active: + break if actual_path: attacker.moved_this_turn = True coords = "->".join(f"({p[0]},{p[1]})" for p in (start, *actual_path)) @@ -1440,6 +1476,8 @@ class CombatEngine: self._withdraw(state, target) elif kind in ("attack", "full_attack"): self._execute_attack(state, target, kind) + elif kind == "immediate": + self._execute_immediate(state, target, action) else: self._execute_movement(state, target, kind) @@ -1476,6 +1514,46 @@ class CombatEngine: elif kind == "wait": self._log(f"round {self._current_round} {state.combatant.id}: wait") + def _execute_immediate( + self, state: CombatantState, target: CombatantState | None, action: Action + ) -> None: + """Execute an immediate action (off-turn or in-turn). + + The immediate action consumes the combatant's next swift action + (``immediate_used``), which suppresses swift/free bonus actions on the + following turn. The effect is resolved by the action's carrier fields: + a spell, a maneuver, a class ability, or a single attack on ``target``. + """ + state.immediate_used = True + if action.spell_name is not None: + self._execute_cast_spell(state, target, action.spell_name) + elif action.maneuver is not None: + self._execute_maneuver(state, target, action.maneuver) + elif action.class_ability is not None: + self._execute_class_ability(state, target, action) + elif target is not None: + self._execute_attack(state, target, "attack") + + def _check_reactions(self, event: ReactionEvent) -> None: + """Offer an event to the reaction policy; execute the first immediate. + + One immediate per event: the first active combatant (in state order) + whose policy returns an action gets to act. Reactions never cascade — + while an immediate is executing, further events are ignored. + """ + if self._reaction_policy is None or self._in_reaction: + return + for state in self._states: + if not state.active or state.immediate_used: + continue + action = self._reaction_policy(self, state, event) + if action is None: + continue + self._in_reaction = True + self._execute(state, action) + self._in_reaction = False + return + def _execute_cast_spell( self, state: CombatantState, @@ -1887,6 +1965,14 @@ class CombatEngine: f"round {self._current_round} {state.combatant.id}: " f"{affected_state.combatant.id} dead" ) + self._check_reactions( + ReactionEvent( + kind="damage_taken", + actor=state, + target=affected_state, + damage=dmg, + ) + ) def _execute_hex( self, state: CombatantState, target: CombatantState | None, hex_name: str | None @@ -2182,6 +2268,10 @@ class CombatEngine: f"round {self._current_round} {caster.combatant.id}: " f"{target.combatant.id} dead" ) + self._check_reactions( + ReactionEvent(kind="spell_cast", actor=caster, target=target) + ) + def _apply_spell_effects( self, spell: SpellSpec, @@ -2246,7 +2336,7 @@ class CombatEngine: if isinstance(effect, DamageEffect): return self._apply_damage_effect( effect, target, caster_live, target_live, - saved=saved, empower=empower, maximize=maximize, + saved=saved, empower=empower, maximize=maximize, caster=caster, ) if isinstance(effect, ConditionEffect): return self._apply_condition_effect(effect, target, saved=saved, spell_name=spell_name) @@ -2264,6 +2354,7 @@ class CombatEngine: saved: bool, empower: bool, maximize: bool, + caster: CombatantState, ) -> str: raw = effect.formula.roll(self._rng) if maximize: @@ -2279,6 +2370,11 @@ class CombatEngine: target.hp -= dmg caster_live.damage_dealt += dmg target_live.damage_taken += dmg + self._check_reactions( + ReactionEvent( + kind="damage_taken", actor=caster, target=target, damage=dmg + ) + ) return f"{dmg} damage" def _apply_condition_effect( @@ -2576,6 +2672,14 @@ class CombatEngine: f"{target.combatant.id} down" ) return True + self._check_reactions( + ReactionEvent( + kind="attack_resolved", + actor=attacker, + target=target, + damage=result.damage, + ) + ) return False def _attack( @@ -2652,6 +2756,9 @@ class CombatEngine: break state.pos = step actual_path.append(step) + self._check_reactions(ReactionEvent(kind="movement_step", actor=state)) + if not state.active: + break if actual_path: state.moved_this_turn = True coords = "->".join(f"({p[0]},{p[1]})" for p in (start, *actual_path)) @@ -2791,6 +2898,9 @@ class CombatEngine: break state.pos = step actual_path.append(step) + self._check_reactions(ReactionEvent(kind="movement_step", actor=state)) + if not state.active: + break if actual_path: state.moved_this_turn = True coords = "->".join(f"({p[0]},{p[1]})" for p in (start, *actual_path)) @@ -2926,6 +3036,16 @@ _FULL_ROUND_KINDS: frozenset[str] = frozenset({"full_attack", "charge", "withdra _STANDARD_KINDS: frozenset[str] = frozenset({ "attack", "cast_spell", "maneuver", "hex", "deed", "spell_combat", "channel_energy", }) +_SWIFT_BONUS_KINDS: frozenset[str] = frozenset({"deed", "inspire_courage", "swift"}) + + +def _without_swift_bonus(actions: tuple[Action, ...]) -> tuple[Action, ...]: + """Drop swift/free bonus actions (deed, inspire_courage, swift). + + An immediate action consumes the combatant's next swift action; the + policy-returned sequence is filtered so no swift bonus fires that turn. + """ + return tuple(a for a in actions if a.kind not in _SWIFT_BONUS_KINDS) def _staggered_filter(actions: tuple[Action, ...]) -> tuple[Action, ...]: @@ -2953,6 +3073,10 @@ def _nauseated_filter(actions: tuple[Action, ...]) -> tuple[Action, ...]: Policy = Callable[[CombatEngine, CombatantState], tuple[Action, ...]] +ReactionPolicy = Callable[ + [CombatEngine, CombatantState, ReactionEvent], Action | None +] + def default_policy(engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]: """Full-attack the nearest enemy; cast spells; charge, 5ft-step, or approach if out of range. diff --git a/tests/test_immediate_actions.py b/tests/test_immediate_actions.py new file mode 100644 index 0000000..eeeb5b7 --- /dev/null +++ b/tests/test_immediate_actions.py @@ -0,0 +1,353 @@ +"""Tests for off-turn immediate actions (B5.4). + +A ``reaction_policy`` hook lets a combatant act outside its own turn with an +``Action(kind="immediate")``. Per PF1e, an immediate action consumes the swift +action of the combatant's next turn. The hook is opt-in: without a +``reaction_policy`` the engine behaves exactly as before. +""" + +from __future__ import annotations + +from typing import Literal + +from pf1e_simulator.combat import ( + Action, + CombatantState, + CombatEngine, + CombatResult, + ReactionEvent, + ReactionPolicy, + default_policy, +) +from pf1e_simulator.dice import parse_dice +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 SeededRng +from pf1e_simulator.spells import DamageEffect, SpellSpec + + +def make_combatant( + cid: str, + *, + hp: int = 30, + ac: int = 15, + attack_bonus: int = 5, + damage: str = "1d6", + initiative_mod: int = 0, + speed: int = 30, + kind: Literal["melee", "ranged"] = "melee", +) -> Combatant: + attack = AttackSpec( + id=f"{cid}-w", + name="short bow" if kind == "ranged" else "short sword", + kind=kind, + attack_bonus=attack_bonus, + damage=[ + DamageComponent( + formula=parse_dice(damage), + types=["piercing"] if kind == "ranged" else ["slashing"], + ) + ], + range_increment_ft=60 if kind == "ranged" else None, + ) + return Combatant( + id=cid, + name=cid, + level=1, + size="Medium", + 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=initiative_mod, + 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() -> Grid: + legend = {".": TerrainType(type="floor", move_cost=1)} + spec = MapSpec(name="test", terrain=tuple(["." * 8] * 8), legend=legend) + return Grid.from_spec(spec) + + +def test_reaction_policy_records_events() -> None: + """A reaction policy is invoked for movement and attack events.""" + hero = make_combatant("hero", initiative_mod=5) + goblin = make_combatant("goblin") + states = [make_state(hero, "players", (1, 1)), make_state(goblin, "monsters", (1, 6))] + events: list[ReactionEvent] = [] + + def approach_policy( + _engine: CombatEngine, state: CombatantState + ) -> tuple[Action, ...]: + if state.combatant.id == "hero": + return ( + Action(kind="move", target_id="goblin"), + Action(kind="attack", target_id="goblin"), + ) + return default_policy(_engine, state) + + def recording_policy( + _engine: CombatEngine, + _state: CombatantState, + event: ReactionEvent, + ) -> Action | None: + events.append(event) + return None + + engine = CombatEngine( + SeededRng(3), + make_grid(), + states, + policy=approach_policy, + reaction_policy=recording_policy, + ) + engine.run() + assert any(e.kind == "movement_step" for e in events) + assert any(e.kind == "attack_resolved" for e in events) + + +def test_no_reaction_policy_runs_unchanged() -> None: + """A no-op reaction policy leaves the transcript bit-identical.""" + + def make_run(reaction_policy: ReactionPolicy | None = None) -> CombatResult: + states = [ + make_state(make_combatant("hero", initiative_mod=5), "players", (1, 1)), + make_state(make_combatant("goblin"), "monsters", (1, 2)), + ] + return CombatEngine( + SeededRng(9), make_grid(), states, reaction_policy=reaction_policy + ).run() + + def noop_policy( + _engine: CombatEngine, + _state: CombatantState, + _event: ReactionEvent, + ) -> Action | None: + return None + + assert make_run(reaction_policy=noop_policy).transcript == make_run().transcript + + +def test_immediate_attack_reaction_executes() -> None: + """A waiting ally can react to an enemy attack with an immediate attack.""" + hero = make_combatant("hero", initiative_mod=5) + goblin = make_combatant("goblin") + hero_state = make_state(hero, "players", (1, 1)) + goblin_state = make_state(goblin, "monsters", (1, 2)) + states = [hero_state, goblin_state] + + def wait_policy( + _engine: CombatEngine, state: CombatantState + ) -> tuple[Action, ...]: + if state.combatant.id == "hero": + return (Action(kind="wait"),) + return default_policy(_engine, state) + + def reaction_policy( + _engine: CombatEngine, state: CombatantState, event: ReactionEvent + ) -> Action | None: + if ( + state.combatant.id == "hero" + and event.kind == "attack_resolved" + and event.target is not None + and event.target.combatant.id == "hero" + ): + return Action(kind="immediate", target_id="goblin") + return None + + engine = CombatEngine( + SeededRng(4), + make_grid(), + states, + policy=wait_policy, + reaction_policy=reaction_policy, + ) + result = engine.run() + assert any("hero: short sword vs goblin" in line for line in result.transcript) + assert goblin_state.hp < 30 + + +def test_immediate_consumes_next_swift() -> None: + """An immediate action consumes the swift action of the next turn.""" + hero = make_combatant("hero", initiative_mod=5) + goblin = make_combatant("goblin") + hero_state = make_state(hero, "players", (1, 1)) + hero_state.resources_left = {"grit": 2} + states = [hero_state, make_state(goblin, "monsters", (1, 2))] + reacted = False + + def deed_policy( + _engine: CombatEngine, state: CombatantState + ) -> tuple[Action, ...]: + if state.combatant.id == "hero": + if state.resources_left.get("grit", 0) > 0: + return ( + Action( + kind="deed", + target_id="goblin", + deed_name="up_close_and_deadly", + ), + Action(kind="attack", target_id="goblin"), + ) + return (Action(kind="attack", target_id="goblin"),) + return default_policy(_engine, state) + + def reaction_policy( + _engine: CombatEngine, state: CombatantState, event: ReactionEvent + ) -> Action | None: + nonlocal reacted + if state.combatant.id == "hero" and event.kind == "attack_resolved" and not reacted: + reacted = True + return Action(kind="immediate", target_id="goblin") + return None + + engine = CombatEngine( + SeededRng(4), + make_grid(), + states, + policy=deed_policy, + reaction_policy=reaction_policy, + ) + result = engine.run() + deed_rounds = [ + int(line.split()[1]) + for line in result.transcript + if "hero: deed Up Close and Deadly activated" in line + ] + assert deed_rounds == [1, 3] + + +def test_immediate_in_turn_sequence_executes() -> None: + """An immediate action in the turn sequence executes like an attack.""" + hero = make_combatant("hero", initiative_mod=5) + goblin = make_combatant("goblin") + hero_state = make_state(hero, "players", (1, 1)) + goblin_state = make_state(goblin, "monsters", (1, 2)) + states = [hero_state, goblin_state] + + def immediate_policy( + _engine: CombatEngine, state: CombatantState + ) -> tuple[Action, ...]: + if state.combatant.id == "hero": + return (Action(kind="immediate", target_id="goblin"),) + return default_policy(_engine, state) + + engine = CombatEngine( + SeededRng(4), + make_grid(), + states, + policy=immediate_policy, + ) + result = engine.run() + assert any("hero: short sword vs goblin" in line for line in result.transcript) + assert goblin_state.hp < 30 + + +def test_reaction_determinism_same_seed_identical_transcripts() -> None: + """Reactions draw from the engine RNG: same seed, identical transcripts.""" + + def make_run() -> CombatResult: + hero = make_combatant("hero", initiative_mod=5) + goblin = make_combatant("goblin") + states = [make_state(hero, "players", (1, 1)), make_state(goblin, "monsters", (1, 2))] + + def wait_policy( + _engine: CombatEngine, state: CombatantState + ) -> tuple[Action, ...]: + if state.combatant.id == "hero": + return (Action(kind="wait"),) + return default_policy(_engine, state) + + def reaction_policy( + _engine: CombatEngine, state: CombatantState, event: ReactionEvent + ) -> Action | None: + if ( + state.combatant.id == "hero" + and event.kind == "attack_resolved" + and event.target is not None + and event.target.combatant.id == "hero" + ): + return Action(kind="immediate", target_id="goblin") + return None + + return CombatEngine( + SeededRng(7), + make_grid(), + states, + policy=wait_policy, + reaction_policy=reaction_policy, + ).run() + + result_one = make_run() + result_two = make_run() + assert result_one.transcript == result_two.transcript + assert result_one.transcript + assert any("hero: short sword vs goblin" in line for line in result_one.transcript) + + +def test_damage_taken_reaction_fires() -> None: + """A reaction policy can react to spell damage taken.""" + caster = make_combatant("caster", hp=50) + reactor = make_combatant("reactor", hp=50, kind="ranged") + caster_state = make_state(caster, "players", (1, 1)) + reactor_state = make_state(reactor, "monsters", (1, 3)) + states = [caster_state, reactor_state] + + bolt = SpellSpec( + name="Bolt", + level=1, + school="evocation", + range="close", + save="none", + effects=(DamageEffect(formula=parse_dice("1d6"), types=("force",)),), + ) + + def cast_policy( + _engine: CombatEngine, state: CombatantState + ) -> tuple[Action, ...]: + if state.combatant.id == "caster": + return (Action(kind="cast_spell", target_id="reactor", spell_name="Bolt"),) + if state.combatant.id == "reactor": + return (Action(kind="wait"),) + return default_policy(_engine, state) + + def reaction_policy( + _engine: CombatEngine, state: CombatantState, event: ReactionEvent + ) -> Action | None: + if state.combatant.id == "reactor" and event.kind == "damage_taken": + return Action(kind="immediate", target_id="caster") + return None + + engine = CombatEngine( + SeededRng(4), + make_grid(), + states, + policy=cast_policy, + spell_registry={"Bolt": bolt}, + reaction_policy=reaction_policy, + ) + result = engine.run() + assert any("reactor: short bow vs caster" in line for line in result.transcript) + assert caster_state.hp < 50