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)


# ── (g) S4b: a readied action changes the readier's initiative (SRD "Ready") ──


def test_readied_action_shifts_initiative_and_persists() -> None:
    # Given: hero (init 10) readies, mid (init 8) waits, foe (init 5) moves
    # When: the foe's round-1 move fires hero's readied attack off-turn
    # Then: hero's initiative becomes the trigger's count + 0.5 (SRD "Ready —
    # Initiative Consequences of Readying": "your initiative result is the
    # count on which you took the readied action, and you act immediately ahead
    # of the character whose action triggered your readied action") — hero drops
    # from first (10) to just ahead of the foe (5.5), behind mid (8), and the
    # new order persists into round 2.
    hero = make_state(_combatant("hero"), "players", (0, 0))
    mid = make_state(_combatant("mid"), "monsters", (5, 5))
    foe = make_state(_combatant("foe"), "monsters", (0, 2))
    eng = _engine(
        # init hero 10, mid 8, foe 5; readied attack d20=10 HIT, dmg 3.
        [10, 8, 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, mid, foe],
    )
    result = eng.run()
    t = result.transcript

    assert "round 1 hero: ready fires (enemy_enters_reach)" in t
    # Hero's initiative key is rewritten to the trigger's count + 0.5, keeping
    # its own modifier (0) and prior list index (0).
    assert eng._initiative_keys["hero"] == (5.5, 0, 0)
    # The trigger's own key is untouched.
    assert eng._initiative_keys["foe"] == (5, 0, 2)
    # Persistence: round 2 order is mid (8), hero (5.5), foe (5) — hero fell
    # from first to second, now acting immediately ahead of the foe.
    idx = _order(t, "round 2 mid:", "round 2 hero:", "round 2 foe:")
    assert idx == tuple(sorted(idx))


def test_readied_initiative_shift_is_visible_in_order_snapshot() -> None:
    # Given/When: the same trigger; the readier's initiative moves to just
    # ahead of the trigger's count.
    # Then: the Observation's initiative_order reflects the new position in the
    # following round (mid, hero, foe — hero no longer first).
    hero = make_state(_combatant("hero"), "players", (0, 0))
    mid = make_state(_combatant("mid"), "monsters", (5, 5))
    foe = make_state(_combatant("foe"), "monsters", (0, 2))
    seen: dict[int, tuple[str, ...]] = {}
    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
        if cid == "hero":
            obs = _snapshot(engine, state)
            seen[engine._current_round] = tuple(obs["initiative_order"])
            return _ready_hero() if n == 0 else _WAIT
        if cid == "foe":
            return (Action(kind="move", target_id="hero"),) if n == 0 else _WAIT
        return _WAIT

    eng = CombatEngine(
        ScriptedRng([10, 8, 5, 10, 3]), make_grid(), [hero, mid, foe],
        policy=policy, round_cap=2,
    )
    eng.run()
    # Round 1 order had hero first; round 2 (after the shift) has hero second,
    # immediately ahead of the foe that triggered the readied action.
    assert seen[1] == ("hero", "mid", "foe")
    assert seen[2] == ("mid", "hero", "foe")


def test_untriggered_ready_does_not_shift_initiative() -> None:
    # Control: a readied action that never fires leaves the initiative key
    # unchanged (the shift happens only when the readied action is taken).
    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],
    )
    eng.run()
    assert eng._initiative_keys["hero"] == (10, 0, 0)
    assert eng._initiative_keys["foe"] == (5, 0, 1)