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

feat(combat): add delay action

Browse Source
This commit is contained in:
Thien An
2026-09-10 12:20:33 +02:00
parent c5cfa78f87
commit f83e4b8016
3 changed files with 393 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+12 -1
View File
@@ -297,6 +297,15 @@ Règles modélisées :
suffit à entrer en portée, ou `(move, attack)` sinon. Les actions immédiates
(hors-tour, consomment le prochain swift) sont modélisées via une politique
de réaction opt-in (`reaction_policy`) — voir « Actions immédiates » ci-dessous.
- Retard (`Action(kind="delay")`, action standard) : le combattant renonce à
agir maintenant et agit plus tard dans le **même round** — en fin d'ordre si
`target_id` est absent, sinon juste avant le combattant désigné. La nouvelle
position est encodée dans la clé d'initiative par un total égal à celui de la
cible − 0,5 (ou au minimum des totaux − 0,5 sans cible) et **persiste aux
rounds suivants** ; le reste du tour est perdu, un second retard dans le même
round est rejeté (`already_delayed`), et le drapeau se réinitialise au début
de chaque round. `default_policy` ne choisit pas le retard (disponible via
une politique personnalisée).
- Retour d'exécution des actions : chaque action produit un `ActionResult`
explicite (kind, ok, reason) consigné dans le snapshot (`last_turn_results`
au tour suivant). Les actions invalides sont rejetées et journalisées au lieu
@@ -304,7 +313,9 @@ Règles modélisées :
déclarés mais non implémentés (`swift`, `free`, `full_round` — leur
implémentation générique reste à faire) → `unsupported_kind` ; dégradation
par les filtres staggered/nauseated (ou swift consommé par une immédiate) →
`downgraded:staggered` / `downgraded:nauseated` / `downgraded:immediate_swift`.
`downgraded:staggered` / `downgraded:nauseated` / `downgraded:immediate_swift` ;
retard valide → `ok=True, reason="delayed"`, second retard du même round →
`ok=False, reason="already_delayed"`.
- Attaques multiples par action : une arme avec `count` > 1 (« 2x Talons »)
résout `count` balayages indépendants dans la même action ; les balayages
restants sont perdus si la cible tombe (inconsciente ou morte) en cours de
src/pf1e_simulator/combat.py
+107 -7
View File
@@ -145,6 +145,7 @@ _PRECISE_SHOT_FEATURE = "Precise Shot" # PF1e: feat negating firing-into-melee
_IMPROVED_CRITICAL_FEATURE = "Improved Critical" # PF1e: feat doubling the threat range
_CRIT_RANGE_BASE = 21 # threat range top is 21 - min natural die (20 -> 20-20)
_DYING_DC = 10 # PF1e: DC of the per-round stabilization Constitution check
_DELAY_INITIATIVE_STEP = 0.5 # PF1e: a delay slots 0.5 below its anchor total
_SAVE_DESCRIPTOR_CONDITIONS = frozenset({
"fear", "cold", "electricity", "fey", "divination", "poison", "disease",
"mind_affecting",
@@ -344,6 +345,11 @@ class CombatantState:
# PREVIOUS turn, assigned at the end of each turn so the Observation built
# at the next policy invocation can expose them.
last_action_results: list[ActionResult] = field(default_factory=list)
# Delay action (not in the frozen data model): set when the combatant took
# the delay action this round; delay_before is the id of the combatant it
# chose to act just before (None = end of the round). Reset at round start.
delayed_this_round: bool = False
delay_before: str | None = None
def usable_attacks(self) -> list[AttackSpec]:
return [
@@ -501,7 +507,9 @@ class Action:
AC), ``5foot_step`` is a free action (one square, no AoO, mutually
exclusive with move/charge/withdraw), ``swift`` and ``free`` are minor
actions, ``immediate`` is an off-turn reaction (deferred). ``wait`` is a
no-op. The engine executes a policy-returned sequence per turn.
no-op. ``delay`` is a standard action that forgoes the turn to act later
in the same round (end of order, or just before ``target_id``). The engine
executes a policy-returned sequence per turn.
"""
kind: Literal[
@@ -509,7 +517,7 @@ class Action:
"full_attack", "charge", "withdraw", "5foot_step", "cast_spell",
"maneuver", "channel_energy", "hex", "deed", "spell_combat",
"inspire_courage", "retrieve_weapon", "repair_weapon",
"activate_power_attack",
"activate_power_attack", "delay",
]
target_id: str | None = None
spell_name: str | None = None
@@ -527,11 +535,13 @@ class Action:
# Normalized action-result reason vocabulary. U1 emits ok, unsupported_kind,
# missing_target and the downgraded:* reasons; no_weapon, out_of_range,
# no_line_of_effect, no_slot and insufficient_resource are declared here for
# the deeper execution-feedback package (U6b).
# the deeper execution-feedback package (U6b). U3 adds delayed /
# already_delayed for the delay action.
ACTION_RESULT_REASONS: frozenset[str] = frozenset({
"ok", "unsupported_kind", "missing_target", "no_weapon", "out_of_range",
"no_line_of_effect", "no_slot", "insufficient_resource",
"downgraded:staggered", "downgraded:nauseated",
"delayed", "already_delayed",
})
_REASON_OK = "ok"
_REASON_UNSUPPORTED_KIND = "unsupported_kind"
@@ -539,6 +549,8 @@ _REASON_MISSING_TARGET = "missing_target"
_REASON_DOWNGRADED_STAGGERED = "downgraded:staggered"
_REASON_DOWNGRADED_NAUSEATED = "downgraded:nauseated"
_REASON_DOWNGRADED_IMMEDIATE_SWIFT = "downgraded:immediate_swift"
_REASON_DELAYED = "delayed"
_REASON_ALREADY_DELAYED = "already_delayed"
# Action kinds that are declared but have no engine dispatch (U1 surfaces
# them as explicit rejections instead of silent no-ops).
@@ -648,7 +660,7 @@ class CombatEngine:
self._stats: dict[str, _LiveStats] = {s.combatant.id: _LiveStats() for s in states}
self._current_round = 0
self._aoo_used: dict[str, int] = {}
self._initiative_keys: dict[str, tuple[int, int, int]] = {}
self._initiative_keys: dict[str, tuple[float, int, int]] = {}
self._pending_summons: list[CombatantState] = []
@property
@@ -1005,10 +1017,14 @@ class CombatEngine:
)
actions = kept
before = len(self._transcript)
was_delayed = state.delayed_this_round
for action in actions:
if not state.active:
break
results.append(self._execute(state, action))
if state.delayed_this_round and not was_delayed:
# A fresh delay forfeits the rest of the turn.
break
if self._winner_side() is not None:
break
if state.active and len(self._transcript) == before:
@@ -1112,19 +1128,53 @@ class CombatEngine:
)
def _run_round(self) -> bool:
"""Execute one round; return True if the battle ended."""
"""Execute one round; return True if the battle ended.
The round order is mutable: a combatant that took a fresh delay action
is re-inserted right before its ``delay_before`` target when that
target is still ahead, otherwise at the end of the round.
"""
for state in self._states:
state.delayed_this_round = False
order = self._build_round_order()
for state in order:
i = 0
while i < len(order):
state = order[i]
if not state.active:
self._process_dying(state)
i += 1
continue
was_delayed = state.delayed_this_round
if self._take_turn(state):
return True
if self._drive_pending_summons():
return True
i += 1
if (
state.delayed_this_round
and not was_delayed
and state not in order[i:]
):
self._reinsert_delayed(order, i, state)
self._pending_summons.clear()
return self._winner_side() is not None
def _reinsert_delayed(
self, order: list[CombatantState], start: int, state: CombatantState
) -> None:
"""Put a freshly-delayed combatant back into the round order.
Inserted immediately before its ``delay_before`` target if that target
still has to act (search from ``start``); appended at the end otherwise.
"""
insert_at = len(order)
if state.delay_before is not None:
for j in range(start, len(order)):
if order[j].combatant.id == state.delay_before:
insert_at = j
break
order.insert(insert_at, state)
def _drive_pending_summons(self) -> bool:
"""Give summoned creatures their turn immediately; return True if battle ended."""
while self._pending_summons:
@@ -1874,7 +1924,7 @@ class CombatEngine:
stable sort.
"""
def sort_key(s: CombatantState) -> tuple[int, int, int]:
def sort_key(s: CombatantState) -> tuple[float, int, int]:
key = self._initiative_keys.get(s.combatant.id)
if key is not None:
return key
@@ -2334,9 +2384,59 @@ class CombatEngine:
f"{kind} rejected ({_REASON_MISSING_TARGET})"
)
return ActionResult(kind=kind, ok=False, reason=_REASON_MISSING_TARGET)
if kind == "delay":
return self._execute_delay(state, target, action)
self._dispatch_action(state, target, action)
return ActionResult(kind=kind, ok=True, reason=_REASON_OK)
def _execute_delay(
self, state: CombatantState, target: CombatantState | None, action: Action
) -> ActionResult:
"""Execute the delay action (PF1e): forgo this turn, act later this round.
With no target the combatant acts at the end of the round; with one it
acts immediately before it. The new initiative total (target total
minus 0.5, or the minimum total minus 0.5) is written into
``_initiative_keys`` so ``_build_round_order`` keeps the position for
the following rounds. A second delay in the same round is rejected
(``already_delayed``) to prevent an infinite delay loop.
"""
kind = action.kind
if state.delayed_this_round:
self._log(
f"round {self._current_round} {state.combatant.id}: "
f"{kind} rejected ({_REASON_ALREADY_DELAYED})"
)
return ActionResult(
kind=kind, ok=False, reason=_REASON_ALREADY_DELAYED
)
key = self._initiative_keys.get(state.combatant.id)
prior_index = key[2] if key is not None else self._states.index(state)
if target is not None:
target_key = self._initiative_keys.get(target.combatant.id)
target_total = target_key[0] if target_key is not None else 0.0
new_total = target_total - _DELAY_INITIATIVE_STEP
else:
totals = [k[0] for k in self._initiative_keys.values()]
new_total = (
min(totals) - _DELAY_INITIATIVE_STEP if totals else 0.0
)
self._initiative_keys[state.combatant.id] = (
new_total,
state.combatant.initiative_mod,
prior_index,
)
state.delayed_this_round = True
state.delay_before = target.combatant.id if target is not None else None
if target is not None:
self._log(
f"round {self._current_round} {state.combatant.id}: "
f"{kind} before {target.combatant.id}"
)
else:
self._log(f"round {self._current_round} {state.combatant.id}: {kind}")
return ActionResult(kind=kind, ok=True, reason=_REASON_DELAYED)
def _dispatch_action(
self, state: CombatantState, target: CombatantState | None, action: Action
) -> None:
tests/test_delay_action.py
+274
View File
@@ -0,0 +1,274 @@
"""The delay action (Package U3).
A combatant that forgoes acting now acts later in the SAME round: with no
target at the end of the round, with ``target_id`` immediately before that
combatant. The new initiative total (anchor - 0.5) persists for the rounds
that follow. A second delay in the same round is rejected (already_delayed),
and ``delayed_this_round`` resets at the start of each round.
"""
from __future__ import annotations
from collections.abc import Callable
from pf1e_simulator.combat import (
Action,
ActionResult,
CombatantState,
CombatEngine,
)
from pf1e_simulator.dice import parse_dice
from pf1e_simulator.models import (
AbilityScores,
ACProfile,
AttackSpec,
Combatant,
DamageComponent,
Saves,
)
from pf1e_simulator.rng import ScriptedRng
from tests.test_observation_enriched import _snapshot, make_grid, make_state
def _combatant(id_: str, *, initiative_mod: int = 0) -> Combatant:
return Combatant(
id=id_,
name=id_,
level=1,
abilities=AbilityScores(
str_score=10, dex_score=10, con_score=10,
int_score=10, wis_score=10, cha_score=10,
),
ac=ACProfile(total=10, touch=10, flat_footed=10),
bab=1,
initiative_mod=initiative_mod,
speed_land_ft=30,
hp_max=100,
saves=Saves(fort=0, ref=0, will=0),
attacks=[
AttackSpec(
id=f"{id_}-sword",
name="short sword",
kind="melee",
attack_bonus=1,
damage=[
DamageComponent(formula=parse_dice("1d6"), types=["slashing"])
],
)
],
)
Decide = Callable[[str, int], tuple[Action, ...]]
PolicyFn = Callable[[CombatEngine, CombatantState], tuple[Action, ...]]
def _counting_policy(
decide: Decide, recorded: list[tuple[str, int, tuple[ActionResult, ...]]]
) -> PolicyFn:
"""Policy calling ``decide(id, call_no)``; records per-call feedback state."""
calls: dict[str, int] = {}
def policy(_engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]:
cid = state.combatant.id
n = calls.get(cid, 0)
calls[cid] = n + 1
recorded.append((cid, n, tuple(state.last_action_results)))
return decide(cid, n)
return policy
def _trio(policy: PolicyFn) -> CombatEngine:
"""hero (mod +2, d20 5 → 7) vs v1 (mod +1, d20 4 → 5) vs v2 (mod 0, d20 3 → 3).
ScriptedRng is fed exactly the three initiative rolls: any other roll the
engine attempts would exhaust the queue and fail the test.
"""
hero = make_state(_combatant("hero", initiative_mod=2), "players", (0, 0))
v1 = make_state(_combatant("v1", initiative_mod=1), "monsters", (0, 3))
v2 = make_state(_combatant("v2"), "monsters", (0, 5))
return CombatEngine(
ScriptedRng([5, 4, 3]), make_grid(), [hero, v1, v2],
policy=policy, round_cap=2,
)
_WAIT: tuple[Action, ...] = (Action(kind="wait"),)
def _order(transcript: tuple[str, ...], *lines: str) -> tuple[int, ...]:
return tuple(transcript.index(line) for line in lines)
# ── (a) delay with no target → acts last this round, position persists ───────
def test_delay_without_target_acts_last_and_persists() -> None:
recorded: list[tuple[str, int, tuple[ActionResult, ...]]] = []
eng = _trio(
_counting_policy(
lambda cid, n: (Action(kind="delay"),)
if (cid, n) == ("hero", 0)
else _WAIT,
recorded,
)
)
result = eng.run()
t = result.transcript
assert "round 1 hero: delay" in t
# Re-queued AFTER everyone else in the same round.
assert t.count("round 1 hero: wait") == 1
idx = _order(
t,
"round 1 hero: delay",
"round 1 v1: wait",
"round 1 v2: wait",
"round 1 hero: wait",
)
assert idx == tuple(sorted(idx))
# The feedback: an ok delay, then the re-queued turn's wait.
hero_results = [r for c, _n, r in recorded if c == "hero"]
assert hero_results[0] == ()
assert hero_results[1] == (
ActionResult(kind="delay", ok=True, reason="delayed"),
)
# Initiative key moved to min total (v2's 3) - 0.5, mod/index kept.
assert eng._initiative_keys["hero"] == (2.5, 2, 0)
# Next rounds act with hero last.
assert _snapshot(eng, eng._states[0])["initiative_order"] == ("v1", "v2", "hero")
assert "round 2 v2: wait" in t
idx2 = _order(t, "round 2 v1: wait", "round 2 v2: wait", "round 2 hero: wait")
assert idx2 == tuple(sorted(idx2))
# ── (b) delay with target_id → inserted just before that combatant ───────────
def test_delay_before_target_inserts_just_before_it() -> None:
recorded: list[tuple[str, int, tuple[ActionResult, ...]]] = []
eng = _trio(
_counting_policy(
lambda cid, n: (Action(kind="delay", target_id="v2"),)
if (cid, n) == ("hero", 0)
else _WAIT,
recorded,
)
)
result = eng.run()
t = result.transcript
assert "round 1 hero: delay before v2" in t
# Order within round 1: hero delays, v1 acts, hero acts, THEN v2 acts.
idx = _order(
t,
"round 1 hero: delay before v2",
"round 1 v1: wait",
"round 1 hero: wait",
"round 1 v2: wait",
)
assert idx == tuple(sorted(idx))
assert [r for c, _n, r in recorded if c == "hero"][1] == (
ActionResult(kind="delay", ok=True, reason="delayed"),
)
# v2's own feedback confirms it acted after the delayed hero (its wait ok).
v2_results = [r for c, _n, r in recorded if c == "v2"]
assert v2_results[1] == (
ActionResult(kind="wait", ok=True, reason="ok"),
)
# New total = v2's 3 - 0.5; v2's own key is untouched.
assert eng._initiative_keys["hero"] == (2.5, 2, 0)
assert eng._initiative_keys["v2"] == (3, 0, 2)
# ── (c) a second delay in the same round → already_delayed ───────────────────
def test_second_delay_same_round_is_rejected() -> None:
recorded: list[tuple[str, int, tuple[ActionResult, ...]]] = []
eng = _trio(
_counting_policy(
lambda cid, n: (Action(kind="delay"),)
if cid == "hero" and n in (0, 1)
else _WAIT,
recorded,
)
)
result = eng.run()
t = result.transcript
assert "round 1 hero: delay" in t
assert "round 1 hero: delay rejected (already_delayed)" in t
# The re-queued turn acted exactly once more; no third insertion happened.
assert t.count("round 1 hero: delay") == 1
assert t.count("round 1 hero: delay rejected (already_delayed)") == 1
hero_turn_lines = [line for line in t if line.startswith("round 1 hero:")]
assert len(hero_turn_lines) == 2
# The rejected delay is reported with ok=False on the following turn.
hero_results = [r for c, _n, r in recorded if c == "hero"]
assert hero_results[2] == (
ActionResult(kind="delay", ok=False, reason="already_delayed"),
)
# The rejected second delay did not move the initiative key again.
assert eng._initiative_keys["hero"] == (2.5, 2, 0)
# ── (d) a round without any delay keeps the pinned transcript byte-identical ─
def test_non_delay_round_transcript_is_pinned() -> None:
recorded: list[tuple[str, int, tuple[ActionResult, ...]]] = []
eng = _trio(_counting_policy(lambda _cid, _n: _WAIT, recorded))
result = eng.run()
assert result.transcript == (
"initiative: hero d20=5+2=7",
"initiative: v1 d20=4+1=5",
"initiative: v2 d20=3+0=3",
"round 1 hero: wait",
"round 1 v1: wait",
"round 1 v2: wait",
"round 2 hero: wait",
"round 2 v1: wait",
"round 2 v2: wait",
"battle over: draw after 2 rounds",
)
assert result.winner is None
assert result.rounds == 2
for state in eng._states:
assert state.delayed_this_round is False
assert state.delay_before is None
# ── (e) delayed_this_round resets at the start of the next round ─────────────
def test_delay_flag_resets_next_round() -> None:
recorded: list[tuple[str, int, tuple[ActionResult, ...]]] = []
eng = _trio(
_counting_policy(
lambda cid, n: (Action(kind="delay"),)
if cid == "hero" and n in (0, 2)
else _WAIT,
recorded,
)
)
result = eng.run()
t = result.transcript
# Hero delays in BOTH rounds — the round-2 delay is accepted, not rejected.
assert "round 1 hero: delay" in t
assert "round 2 hero: delay" in t
assert not any("already_delayed" in line for line in t)
# Round 2: hero (now last, total 2.5) delays again → min 2.5 - 0.5 = 2.0.
idx = _order(
t,
"round 2 v1: wait",
"round 2 v2: wait",
"round 2 hero: delay",
"round 2 hero: wait",
)
assert idx == tuple(sorted(idx))
assert eng._initiative_keys["hero"] == (2.0, 2, 0)
hero = eng._states[0]
assert hero.delayed_this_round is True