pf1e-simulator/tests/test_delay_action.py

"""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.effects import ActiveBuff, StatModifier
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


# ── (f) a delayed turn does not double-process the start of the round ─────────


def _delay_bleed_engine() -> CombatEngine:
    """hero (mod +2 → 7, delays then acts) vs v1 (mod +1 → 5) vs v2 (mod 0 → 3).

    hero is bleeding and carries a 2-round buff; the initiative rolls are the
    only d20s the engine may consume (a repeated dying/confusion roll or a
    second bleed tick that drops hero below 0 would raise on an empty queue).
    """
    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))
    eng = CombatEngine(
        ScriptedRng([7, 5, 3]), make_grid(), [hero, v1, v2],
        policy=_counting_policy(
            lambda cid, n: (Action(kind="delay"),) if (cid, n) == ("hero", 0) else _WAIT,
            [],
        ),
        round_cap=1,
    )
    hero.bleed_damage = 1
    hero.active_buffs.append(
        ActiveBuff(modifiers=(StatModifier(target="ac", value=2, bonus_type="insight"),),
                   rounds_left=2, source="test")
    )
    return eng


def _bleed_ticks(eng: CombatEngine, cid: str) -> int:
    return sum(1 for ln in eng._transcript if f"{cid}: bleed" in ln)


def test_delay_does_not_double_bleed_tick() -> None:
    # Given: a bleeding combatant that delays; When: it acts later the same
    # round; Then: bleed ticks exactly once (the re-inserted turn skips the
    # start-of-turn gate).
    eng = _delay_bleed_engine()
    eng.run()
    assert _bleed_ticks(eng, "hero") == 1


def test_delay_does_not_double_decrement_buff_duration() -> None:
    # Given: a delayed combatant carrying a 2-round buff; When: the round runs
    # (hero acts twice: the delay turn and the re-inserted turn); Then: the
    # buff duration decrements once (2 -> 1), not twice.
    eng = _delay_bleed_engine()
    hero = eng._states[0]
    eng.run()
    assert [b.rounds_left for b in hero.active_buffs] == [1]