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

feat(combat): support off-turn immediate actions via reaction policy

Browse Source
This commit is contained in:
Thien An
2026-09-05 17:03:51 +02:00
parent ab99ab67e3
commit 66b6d1f387
2 changed files with 481 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pf1e_simulator/combat.py
+128 -4
View File
@@ -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.
tests/test_immediate_actions.py
+353
View File
@@ -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