pf1e-simulator/tests/test_ready_action.py

"""The ready action (Package U4).

A combatant spends its standard action preparing a payload (attack, cast_spell
or maneuver, encoded via the Action carrier fields like the immediate action).
When a matching ``ReactionEvent`` from an ENEMY occurs, the engine fires the
payload off-turn through the reaction machinery — no swift cost — and the
readied action is consumed (one-shot). Triggers: ``enemy_enters_reach`` ←
movement_step, ``enemy_casts`` ← spell_cast, ``enemy_attacks`` ←
attack_resolved, ``enemy_hits_ally`` ← damage_taken.
"""
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 pf1e_simulator.spells import DamageEffect, SpellSpec
from tests.test_observation_enriched import _snapshot, make_grid, make_state


def _combatant(id_: str, *, spells: tuple[str, ...] = ()) -> 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=0,
        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=5,
                damage=[
                    DamageComponent(formula=parse_dice("1d6"), types=["slashing"])
                ],
            )
        ],
        spells=list(spells),
    )


_BOLT = SpellSpec(
    name="Bolt",
    level=1,
    school="evocation",
    range="close",
    save="none",
    effects=(DamageEffect(formula=parse_dice("1d6"), types=("force",)),),
)

PolicyFn = Callable[[CombatEngine, CombatantState], tuple[Action, ...]]

_WAIT: tuple[Action, ...] = (Action(kind="wait"),)


def _counting(decide: Callable[[str, int], tuple[Action, ...]]) -> PolicyFn:
    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
        return decide(cid, n)

    return policy


def _engine(rng: list[int], policy: PolicyFn, states: list[CombatantState]) -> CombatEngine:
    return CombatEngine(ScriptedRng(rng), make_grid(), states, policy=policy, round_cap=2)


def _ready_hero() -> tuple[Action, ...]:
    return (
        Action(kind="ready", ready_trigger="enemy_enters_reach", target_id="foe"),
    )


def _order(transcript: tuple[str, ...], *needles: str) -> tuple[int, ...]:
    """Indices of the first transcript line containing each needle."""
    return tuple(
        next(i for i, line in enumerate(transcript) if needle in line)
        for needle in needles
    )


# ── (a) readied attack fires when an enemy moves (movement_step) ──────────────


def test_readied_attack_fires_on_enemy_movement() -> None:
    hero = make_state(_combatant("hero"), "players", (0, 0))
    foe = make_state(_combatant("foe"), "monsters", (0, 2))
    eng = _engine(
        # init hero 10, init foe 5, readied attack d20=10 (+5 ≥ AC 10 HIT), dmg 3.
        [10, 5, 10, 3],
        _counting(
            lambda cid, n: (
                _ready_hero()
                if (cid, n) == ("hero", 0)
                else (Action(kind="move", target_id="hero"),)
                if (cid, n) == ("foe", 0)
                else _WAIT
            )
        ),
        [hero, foe],
    )
    result = eng.run()
    t = result.transcript

    assert "round 1 hero: ready (enemy_enters_reach)" in t
    assert "round 1 hero: ready fires (enemy_enters_reach)" in t
    attack = "round 1 hero: short sword vs foe d20=10+5=15 AC 10 -> HIT 3 damage (100->97)"
    assert attack in t
    # Off-turn: the attack resolves DURING the foe's move, before its log line.
    idx = _order(
        t,
        "round 1 hero: ready (enemy_enters_reach)",
        "round 1 hero: ready fires (enemy_enters_reach)",
        attack,
        "round 1 foe: move (0,2)->(0,1)",
    )
    assert idx == tuple(sorted(idx))
    # Consumed: the readied action is gone after firing.
    assert hero.readied is None
    assert hero.readied_trigger is None
    assert foe.hp == 97


# ── (b) readied cast_spell fires on an enemy spell_cast ───────────────────────


def test_readied_cast_fires_on_enemy_spell_cast() -> None:
    hero = make_state(_combatant("hero", spells=("Bolt",)), "players", (0, 0))
    hero.spell_slots_left = {1: 1}
    foe = make_state(_combatant("foe", spells=("Bolt",)), "monsters", (0, 2))
    foe.spell_slots_left = {1: 1}
    eng = CombatEngine(
        # init hero 10, init foe 5, foe's Bolt dmg 4, readied Bolt dmg 2.
        ScriptedRng([10, 5, 4, 2]),
        make_grid(),
        [hero, foe],
        spell_registry={"Bolt": _BOLT},
        policy=_counting(
            lambda cid, n: (
                (
                    Action(
                        kind="ready",
                        ready_trigger="enemy_casts",
                        target_id="foe",
                        spell_name="Bolt",
                    ),
                )
                if (cid, n) == ("hero", 0)
                else (Action(kind="cast_spell", target_id="hero", spell_name="Bolt"),)
                if (cid, n) == ("foe", 0)
                else _WAIT
            )
        ),
        round_cap=2,
    )
    result = eng.run()
    t = result.transcript

    assert "round 1 hero: ready (enemy_casts)" in t
    assert any("round 1 foe: cast Bolt vs hero" in line for line in t)
    assert "round 1 hero: ready fires (enemy_casts)" in t
    assert any("round 1 hero: cast Bolt vs foe" in line for line in t)
    idx = _order(
        t,
        "round 1 foe: cast Bolt vs hero",
        "round 1 hero: ready fires (enemy_casts)",
        "round 1 hero: cast Bolt vs foe",
    )
    assert idx == tuple(sorted(idx))
    assert hero.hp == 96
    assert foe.hp == 98
    # The readied cast spent hero's own slot and was consumed.
    assert hero.spell_slots_left == {1: 0}
    assert hero.readied is None


# ── (c) no fire on a non-matching trigger nor on a friendly event ─────────────


def test_readied_ignores_non_matching_trigger() -> None:
    hero = make_state(_combatant("hero"), "players", (0, 0))
    foe = make_state(_combatant("foe"), "monsters", (0, 2))
    eng = _engine(
        # init hero 10, init foe 5, foe's attack d20=2 (+5=7 < AC 10 MISS).
        [10, 5, 2],
        _counting(
            lambda cid, n: (
                (
                    Action(
                        kind="ready",
                        ready_trigger="enemy_casts",
                        target_id="foe",
                    ),
                )
                if (cid, n) == ("hero", 0)
                else (
                    Action(kind="move", target_id="hero"),
                    Action(kind="attack", target_id="hero"),
                )
                if (cid, n) == ("foe", 0)
                else _WAIT
            )
        ),
        [hero, foe],
    )
    result = eng.run()
    t = result.transcript

    # movement_step AND attack_resolved: neither matches enemy_casts.
    assert "round 1 foe: move (0,2)->(0,1)" in t
    assert "round 1 foe: short sword vs hero d20=2+5=7 AC 10 -> MISS" in t
    assert not any("ready fires" in line for line in t)
    # SRD "Ready": the trigger never occurred before the hero's next action
    # (round 2), so the readied action expires at the start of that turn.
    assert any("ready expired" in line for line in t)
    assert hero.readied is None
    assert hero.readied_trigger is None


def test_readied_ignores_friendly_movement() -> None:
    hero = make_state(_combatant("hero"), "players", (0, 0))
    ally = make_state(_combatant("ally"), "players", (2, 0))
    foe = make_state(_combatant("foe"), "monsters", (0, 4))
    eng = _engine(
        [10, 9, 5],  # initiative only — nothing ever attacks.
        _counting(
            lambda cid, n: (
                _ready_hero()
                if (cid, n) == ("hero", 0)
                else (Action(kind="move", target_id="hero"),)
                if (cid, n) == ("ally", 0)
                else _WAIT
            )
        ),
        [hero, ally, foe],
    )
    result = eng.run()
    t = result.transcript

    # The ally's movement_step is a friendly event: the ready holds through
    # round 1. At the start of the hero's round 2 turn it expires (SRD
    # "Ready": "If you come to your next action and have not yet performed
    # your readied action, you don't get to take the readied action").
    assert "round 1 ally: move (2,0)->(1,0)" in t
    assert not any("ready fires" in line for line in t)
    assert hero.readied is None
    assert any("ready expired" in line for line in t)


# ── (d) the readied action is consumed (one-shot) ─────────────────────────────


def test_readied_action_fires_only_once() -> None:
    hero = make_state(_combatant("hero"), "players", (0, 0))
    v1 = make_state(_combatant("v1"), "monsters", (0, 2))
    v2 = make_state(_combatant("v2"), "monsters", (2, 2))
    eng = _engine(
        # init hero 10, v1 5, v2 4; readied attack on v1: d20=10 HIT, dmg 3.
        [10, 5, 4, 10, 3],
        _counting(
            lambda cid, n: (
                (
                    Action(
                        kind="ready",
                        ready_trigger="enemy_enters_reach",
                        target_id="v1",
                    ),
                )
                if (cid, n) == ("hero", 0)
                else (Action(kind="move", target_id="hero"),)
                if cid in ("v1", "v2") and n == 0
                else _WAIT
            )
        ),
        [hero, v1, v2],
    )
    result = eng.run()
    t = result.transcript

    assert sum(1 for line in t if "ready fires" in line) == 1
    assert "round 1 hero: short sword vs v1 d20=10+5=15 AC 10 -> HIT 3 damage (100->97)" in t
    # v2's later movement_step is a second matching event: no second fire.
    assert "round 1 v2: move (2,2)->(1,1)" in t
    assert not any("vs v2" in line for line in t)
    assert hero.readied is None


# ── (d2) expiry at the combatant's next turn (SRD "Ready") ────────────────────


def test_readied_action_fires_in_the_next_round_before_own_turn() -> None:
    # Given: the foe (init 10) waits in round 1; the hero (init 5) readies;
    #        in round 2 the foe moves before the hero's own turn comes
    # When: the battle runs two rounds
    # Then: the readied action survives the round boundary and fires on the
    # foe's round-2 move (it expires only when the hero's own turn arrives).
    hero = make_state(_combatant("hero"), "players", (0, 0))
    foe = make_state(_combatant("foe"), "monsters", (0, 2))
    eng = _engine(
        # init hero 5, foe 10; round 1: foe wait, hero ready; round 2: foe
        # moves BEFORE the hero's turn and fires the readied attack
        # (d20=10 +5 = 15 >= AC 10, dmg 3).
        [5, 10, 10, 3],
        _counting(
            lambda cid, n: (
                _WAIT
                if cid == "foe" and n == 0
                else (Action(kind="move", target_id="hero"),)
                if cid == "foe"
                else _ready_hero()
                if n == 0
                else _WAIT
            )
        ),
        [hero, foe],
    )
    result = eng.run()
    t = result.transcript

    assert "round 2 foe: move (0,2)->(0,1)" in t
    assert any("round 2" in line and "ready fires (enemy_enters_reach)" in line for line in t)
    assert "round 2 hero: short sword vs foe d20=10+5=15 AC 10 -> HIT 3 damage (100->97)" in t
    assert hero.readied is None
    assert not any("ready expired" in line for line in t)


def test_readied_action_expires_at_start_of_own_turn() -> None:
    # Given: the hero readies with a trigger that never occurs
    # When: the hero's next turn starts
    # Then: the readied action is cleared and logged as expired, and the
    # hero may ready again that turn.
    hero = make_state(_combatant("hero"), "players", (0, 0))
    foe = make_state(_combatant("foe"), "monsters", (0, 6))
    eng = _engine(
        [10, 5],  # initiative only — the foe never comes close.
        _counting(
            lambda cid, _n: (
                _ready_hero()
                if cid == "hero"
                else _WAIT
            )
        ),
        [hero, foe],
    )
    result = eng.run()
    t = result.transcript

    assert sum(1 for line in t if "ready expired" in line) == 1
    assert "round 2 hero: ready expired (trigger never occurred)" in t
    # Round 2's policy re-readies (n == 1 still returns the ready action),
    # so the state is re-armed after the expiry.
    assert hero.readied is not None
    assert hero.readied_trigger == "enemy_enters_reach"


# ── (e) the snapshot exposes readied = {kind, trigger}, None after firing ─────


def test_snapshot_exposes_readied_until_fired() -> None:
    hero = make_state(_combatant("hero"), "players", (0, 0))
    foe = make_state(_combatant("foe"), "monsters", (0, 2))
    eng = _engine(
        [10, 5, 10, 3],
        _counting(
            lambda cid, n: (
                _ready_hero()
                if (cid, n) == ("hero", 0)
                else (Action(kind="move", target_id="hero"),)
                if (cid, n) == ("foe", 0)
                else _WAIT
            )
        ),
        [hero, foe],
    )
    eng._roll_initiative()
    eng._take_turn(hero)

    assert hero.last_action_results == [
        ActionResult(kind="ready", ok=True, reason="ready")
    ]
    snap = _snapshot(eng, hero)["combatants"][0]
    assert snap["readied"] == {"kind": "attack", "trigger": "enemy_enters_reach"}

    eng._take_turn(foe)  # the foe's move fires the readied attack.
    snap_after = _snapshot(eng, hero)["combatants"][0]
    assert snap_after["readied"] is None
    assert hero.readied is None


# ── (f) no cascade: a fired readied action cannot trigger another ready ───────


def test_readied_payload_events_do_not_cascade() -> None:
    hero = make_state(_combatant("hero"), "players", (0, 0))
    v1 = make_state(_combatant("v1"), "monsters", (0, 2))
    eng = _engine(
        # init hero 10, v1 5; the readied attack fires on v1's round-1 move:
        # d20=10 HIT, dmg 3.
        [10, 5, 10, 3],
        _counting(
            lambda cid, n: (
                _ready_hero()
                if (cid, n) == ("hero", 0)
                else (
                    Action(
                        kind="ready",
                        ready_trigger="enemy_attacks",
                        target_id="hero",
                    ),
                    Action(kind="move", target_id="hero"),
                )
                if (cid, n) == ("v1", 0)
                else _WAIT
            )
        ),
        [hero, v1],
    )
    result = eng.run()
    t = result.transcript

    # Round 1: v1's move (its move action, right after its own ready) fires
    # hero's readied attack; the attack_resolved event it generates must NOT
    # fire v1's own readied counter-attack (both readys are still live —
    # neither has come to its next action yet).
    assert "round 1 hero: ready fires (enemy_enters_reach)" in t
    assert sum(1 for line in t if "ready fires" in line) == 1
    assert not any("v1: short sword vs hero" in line for line in t)
    assert hero.hp == 100
    # Round 2: v1's counter-ready expires at the start of v1's own turn
    # (SRD "Ready": "If you come to your next action and have not yet
    # performed your readied action, you don't get to take the readied
    # action").
    assert v1.readied is None
    assert any("round 2 v1: ready expired" in line for line in t)