pf1e-simulator/tests/test_conformance_action_economy.py

"""Conformance X4: the general per-turn action-economy budget.

PF1e CRB "Actions in Combat": one standard action + one move action (or one
full-round action in place of both), plus one swift action per turn; free and
conditional actions never consume the budget. ``_action_economy_filter``
enforces this on EVERY policy-returned sequence (unconditionally, not only
for the immediate/staggered/nauseated specializations): surplus actions are
dropped in order (the earliest legal prefix-subsequence is kept) and reported
as ``downgraded:action_economy`` in ``last_action_results``. ``default_policy``
turns are legal by construction — verified here so the goldens stay frozen.
"""

from __future__ import annotations

from collections.abc import Callable
from typing import Literal

from pf1e_simulator.combat import (
    _ACTION_ECONOMY_FREE_KINDS,
    _ACTION_ECONOMY_FULL_ROUND_KINDS,
    _ACTION_ECONOMY_MOVE_KINDS,
    _ACTION_ECONOMY_STANDARD_KINDS,
    _ACTION_ECONOMY_SWIFT_KINDS,
    ACTION_RESULT_REASONS,
    Action,
    CombatantState,
    CombatEngine,
    _action_economy_filter,
    _action_economy_slot,
    _dropped_kinds,
    _dropped_kinds_multiset,
    default_policy,
)
from pf1e_simulator.conditions import CONDITIONS_BY_NAME
from pf1e_simulator.effects import ActiveCondition
from pf1e_simulator.rng import ScriptedRng
from tests.test_combat import make_combatant, make_state
from tests.test_observation_enriched import make_grid

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

_Kind = Literal[
    "attack", "move", "wait", "swift", "free", "immediate", "full_round",
    "full_attack", "charge", "withdraw", "5foot_step", "cast_spell",
    "maneuver", "channel_energy", "hex", "deed", "spell_combat",
    "inspire_courage", "retrieve_weapon", "repair_weapon",
    "activate_power_attack", "delay", "ready", "total_defense",
    "fight_defensively", "stand_up", "drop_prone", "aid_another", "run",
]

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


def _foe(kind: _Kind) -> Action:
    return Action(kind=kind, target_id="foe")


# ── (a) the kind -> slot classification is exhaustive and disjoint ───────────

_ALL_KINDS: tuple[_Kind, ...] = (
    "attack", "move", "wait", "swift", "free", "immediate", "full_round",
    "full_attack", "charge", "withdraw", "5foot_step", "cast_spell",
    "maneuver", "channel_energy", "hex", "deed", "spell_combat",
    "inspire_courage", "retrieve_weapon", "repair_weapon",
    "activate_power_attack", "delay", "ready", "total_defense",
    "fight_defensively", "stand_up", "drop_prone", "aid_another", "run",
)


def test_every_action_kind_is_classified_exactly_once() -> None:
    groups = (
        _ACTION_ECONOMY_STANDARD_KINDS,
        _ACTION_ECONOMY_MOVE_KINDS,
        _ACTION_ECONOMY_SWIFT_KINDS,
        _ACTION_ECONOMY_FULL_ROUND_KINDS,
        _ACTION_ECONOMY_FREE_KINDS,
    )
    covered = frozenset().union(*groups)
    assert covered == frozenset(_ALL_KINDS)
    # Pairwise disjoint: no kind consumes two slots.
    total = sum(len(g) for g in groups)
    assert total == len(covered)


def test_slot_per_kind_matches_pf1e_action_type() -> None:
    expected: dict[str, str] = {
        # standard actions
        "attack": "standard", "cast_spell": "standard", "maneuver": "standard",
        "channel_energy": "standard", "hex": "standard", "delay": "standard",
        "ready": "standard", "total_defense": "standard",
        "aid_another": "standard",
        # move actions
        "move": "move", "stand_up": "move", "retrieve_weapon": "move",
        "repair_weapon": "move",
        # run: RAW full-round, but the engine models it as move machinery
        # (sets moved_this_turn, excluded from _FULL_ROUND_KINDS) — consistent.
        "run": "move",
        # swift actions
        "swift": "swift", "deed": "swift", "inspire_courage": "swift",
        # full-round actions
        "full_attack": "full_round", "charge": "full_round",
        "withdraw": "full_round", "full_round": "full_round",
        "spell_combat": "full_round",
        # free / conditional — no budget
        "wait": "free", "free": "free", "immediate": "free",
        "5foot_step": "free", "drop_prone": "free",
        "activate_power_attack": "free", "fight_defensively": "free",
    }
    assert set(expected) == set(_ALL_KINDS)
    for kind, slot in expected.items():
        assert _action_economy_slot(kind) == slot, kind


def test_downgrade_reason_is_in_the_result_vocabulary() -> None:
    assert "downgraded:action_economy" in ACTION_RESULT_REASONS


# ── (b) legal sequences pass through untouched ───────────────────────────────


def test_legal_sequences_are_kept_unchanged() -> None:
    legal: tuple[tuple[_Kind, ...], ...] = (
        ("attack", "move"),
        ("move", "attack"),
        ("deed", "attack"),
        ("swift", "attack"),
        ("full_attack",),
        ("charge",),
        ("withdraw",),
        ("run",),
        ("5foot_step", "full_attack"),
        ("move", "5foot_step"),
        ("5foot_step", "move"),
        ("inspire_courage", "spell_combat"),
        ("activate_power_attack", "full_attack"),
        ("drop_prone", "attack"),
        ("fight_defensively", "attack"),
        ("deed", "move", "attack"),
        ("wait",),
    )
    for kinds in legal:
        actions = tuple(_foe(k) for k in kinds)
        assert _action_economy_filter(actions) == actions, kinds
        assert _dropped_kinds_multiset(actions, actions) == (), kinds


def test_two_moves_conflict_but_free_steps_do_not() -> None:
    # stand_up is a move action: a second move is surplus.
    actions = (_foe("move"), _foe("stand_up"))
    assert _action_economy_filter(actions) == (_foe("move"),)
    # 5foot_step is free: two steps do not consume the move slot.
    steps = (_foe("5foot_step"), _foe("5foot_step"))
    assert _action_economy_filter(steps) == steps


# ── (c) illegal sequences degrade to the earliest legal prefix ───────────────


def test_illegal_sequences_degrade_deterministically() -> None:
    cases: tuple[tuple[tuple[_Kind, ...], tuple[_Kind, ...]], ...] = (
        # two standards — the second is surplus
        (("attack", "attack"), ("attack",)),
        # charge (full-round) then a standard attack
        (("charge", "attack"), ("charge",)),
        # standard spent, then a full-round conflicting with standard+move
        (("attack", "full_attack", "move"), ("attack", "move")),
        # two moves around a standard
        (("move", "attack", "move"), ("move", "attack")),
        # two swifts — the second is surplus
        (("full_attack", "deed", "inspire_courage"), ("full_attack", "deed")),
        # full-round then a move (charge consumes standard+move)
        (("full_attack", "move"), ("full_attack",)),
        # two full-round actions
        (("charge", "withdraw"), ("charge",)),
        # move first, then a full-round that needed the move slot
        (("move", "charge"), ("move",)),
        # three standards
        (("attack", "cast_spell", "maneuver"), ("attack",)),
    )
    for kinds, kept_kinds in cases:
        actions = tuple(_foe(k) for k in kinds)
        expected = tuple(_foe(k) for k in kept_kinds)
        assert _action_economy_filter(actions) == expected, kinds


def test_multiset_drop_reports_each_surplus_instance() -> None:
    actions = (_foe("attack"), _foe("attack"))
    assert _dropped_kinds_multiset(actions, (actions[0],)) == ("attack",)
    # The set-based _dropped_kinds cannot see the duplicate; the multiset can.
    assert _dropped_kinds(actions, (actions[0],)) == ()
    deeds = (_foe("deed"), _foe("deed"), _foe("attack"))
    kept = (deeds[0], deeds[2])
    assert _dropped_kinds_multiset(deeds, kept) == ("deed",)


# ── (d) interplay with the condition-specific filters ────────────────────────


def _duel(policy: PolicyFn, rng: list[int], round_cap: int = 1) -> CombatEngine:
    hero = make_state(
        make_combatant("hero", hp=100, initiative_mod=2), "players", (0, 0)
    )
    foe = make_state(make_combatant("foe", hp=100), "monsters", (0, 1))
    return CombatEngine(
        ScriptedRng(rng), make_grid(), [hero, foe],
        policy=policy, round_cap=round_cap,
    )


def _add_condition(state: CombatantState, name: str) -> None:
    state.active_conditions.append(
        ActiveCondition(
            condition=CONDITIONS_BY_NAME[name], rounds_left=5, source="test"
        )
    )


def test_staggered_after_budget_keeps_single_legal_action() -> None:
    # (charge, attack): the budget keeps the charge and drops the attack as
    # surplus standard; the staggered filter then drops the full-round charge.
    def policy(_e: CombatEngine, s: CombatantState) -> tuple[Action, ...]:
        if s.combatant.id == "hero":
            return (
                Action(kind="charge", target_id="foe"),
                Action(kind="attack", target_id="foe"),
            )
        return _WAIT

    eng = _duel(policy, [10, 5])
    hero = eng._states[0]
    hero.hp = 0  # disabled -> staggered
    eng._roll_initiative()
    eng._take_turn(hero)
    reasons = [(r.kind, r.reason) for r in hero.last_action_results]
    assert ("attack", "downgraded:action_economy") in reasons
    assert ("charge", "downgraded:staggered") in reasons


def test_nauseated_after_budget_keeps_single_move() -> None:
    # (move, attack, move): the budget drops the second move; the nauseated
    # filter then keeps only the first move and drops the standard attack.
    def policy(_e: CombatEngine, s: CombatantState) -> tuple[Action, ...]:
        if s.combatant.id == "hero":
            return (
                Action(kind="move", target_id="foe"),
                Action(kind="attack", target_id="foe"),
                Action(kind="move", target_id="foe"),
            )
        return _WAIT

    eng = _duel(policy, [10, 5])
    hero = eng._states[0]
    _add_condition(hero, "nauseated")
    eng._roll_initiative()
    eng._take_turn(hero)
    reasons = [(r.kind, r.reason) for r in hero.last_action_results]
    assert reasons.count(("move", "downgraded:action_economy")) == 1
    assert ("attack", "downgraded:nauseated") in reasons


def test_immediate_swift_still_removes_swift_bonus_after_budget() -> None:
    # (deed, deed, attack): the budget drops the second deed; the consumed
    # swift (immediate_used) then removes the surviving deed too.
    def policy(_e: CombatEngine, s: CombatantState) -> tuple[Action, ...]:
        if s.combatant.id == "hero":
            return (
                Action(
                    kind="deed", target_id="foe",
                    deed_name="up_close_and_deadly",
                ),
                Action(
                    kind="deed", target_id="foe",
                    deed_name="up_close_and_deadly",
                ),
                Action(kind="attack", target_id="foe"),
            )
        return _WAIT

    eng = _duel(policy, [10, 5, 8, 3])
    hero = eng._states[0]
    hero.immediate_used = True
    eng._roll_initiative()
    eng._take_turn(hero)
    reasons = [(r.kind, r.reason) for r in hero.last_action_results]
    assert ("deed", "downgraded:action_economy") in reasons
    assert ("deed", "downgraded:immediate_swift") in reasons
    assert ("attack", "ok") in reasons


# ── (e) end-to-end: an illegal custom policy tuple is degraded by run() ─────


def test_run_degrades_illegal_policy_and_reports_results() -> None:
    def policy(_e: CombatEngine, s: CombatantState) -> tuple[Action, ...]:
        if s.combatant.id == "hero":
            # two standards + two moves: only the first attack and the
            # first move may execute.
            return (
                Action(kind="attack", target_id="foe"),
                Action(kind="attack", target_id="foe"),
                Action(kind="move", target_id="foe"),
                Action(kind="move", target_id="foe"),
            )
        return _WAIT

    eng = _duel(policy, [10, 5, 8, 3, 6, 4])
    result = eng.run()
    hero = eng._states[0]
    reasons = [(r.kind, r.ok, r.reason) for r in hero.last_action_results]
    assert reasons == [
        ("attack", False, "downgraded:action_economy"),
        ("move", False, "downgraded:action_economy"),
        ("attack", True, "ok"),
        ("move", True, "ok"),
    ]
    # Only one hero swing is resolved in the transcript.
    swings = [ln for ln in result.transcript if "hero: short sword vs foe" in ln]
    assert len(swings) == 1


def test_run_keeps_legal_custom_policy_untouched() -> None:
    def policy(_e: CombatEngine, s: CombatantState) -> tuple[Action, ...]:
        if s.combatant.id == "hero":
            return (
                Action(
                    kind="deed", target_id="foe",
                    deed_name="up_close_and_deadly",
                ),
                Action(kind="move", target_id="foe"),
                Action(kind="attack", target_id="foe"),
            )
        return _WAIT

    eng = _duel(policy, [10, 5, 8, 3, 6, 4])
    eng.run()
    hero = eng._states[0]
    for res in hero.last_action_results:
        assert res.reason != "downgraded:action_economy"


# ── (f) default_policy turns are legal by construction (golden invariant) ────


def test_default_policy_representative_sequences_pass_the_validator() -> None:
    shapes: tuple[tuple[_Kind, ...], ...] = (
        ("wait",),
        ("withdraw",),
        ("channel_energy",),
        ("hex",),
        ("cast_spell",),
        ("inspire_courage", "spell_combat"),
        ("deed", "full_attack"),
        ("deed", "charge"),
        ("deed", "5foot_step", "full_attack"),
        ("deed", "move", "attack"),
        ("inspire_courage", "cast_spell"),
        ("full_attack",),
        ("charge",),
        ("5foot_step", "full_attack"),
        ("move", "attack"),
    )
    for kinds in shapes:
        actions = tuple(_foe(k) for k in kinds)
        assert _action_economy_filter(actions) == actions, kinds


def test_default_policy_combats_never_trigger_the_budget_downgrade() -> None:
    # All d20 rolls are 1s after initiative: every swing auto-misses, so the
    # scripted queue cannot exhaust before the round cap.
    for rng in (
        [10, 5] + [1] * 10,
        [8, 6] + [1] * 10,
    ):
        eng = _duel(default_policy, rng, round_cap=3)
        eng.run()
        for state in eng._states:
            for res in state.last_action_results:
                assert res.reason != "downgraded:action_economy", (rng, res)