diff --git a/README.md b/README.md index eb2c508..d140e09 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/src/pf1e_simulator/combat.py b/src/pf1e_simulator/combat.py index f946a73..0cc0983 100644 --- a/src/pf1e_simulator/combat.py +++ b/src/pf1e_simulator/combat.py @@ -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: diff --git a/tests/test_delay_action.py b/tests/test_delay_action.py new file mode 100644 index 0000000..fc9100e --- /dev/null +++ b/tests/test_delay_action.py @@ -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