pf1e-simulator/tests/test_action_feedback.py

"""Explicit action-execution feedback (Package U1).

Every Action produces an ActionResult; the previous turn's results are exposed
on CombatantState.last_action_results and in the snapshot under the additive
per-combatant key ``last_turn_results`` — so an LLM policy can learn which
actions were rejected (missing target, unsupported kind) or downgraded by the
staggered/nauseated filters.
"""
from __future__ import annotations

import json
from collections.abc import Callable

from pf1e_simulator.combat import (
    Action,
    ActionResult,
    CombatantState,
    CombatEngine,
    LLMPolicy,
    Observation,
)
from pf1e_simulator.conditions import CONDITIONS_BY_NAME
from pf1e_simulator.dice import parse_dice
from pf1e_simulator.effects import ActiveCondition
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 _melee(name: str, attack_bonus: int = 5) -> AttackSpec:
    return AttackSpec(
        id=f"{name}-sword",
        name="short sword",
        kind="melee",
        attack_bonus=attack_bonus,
        damage=[DamageComponent(formula=parse_dice("1d6"), types=["slashing"])],
    )


def _combatant(id_: str, *, ac: int = 10) -> 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=ac, touch=ac, flat_footed=ac),
        bab=1,
        initiative_mod=0,
        speed_land_ft=30,
        hp_max=100,
        saves=Saves(fort=0, ref=0, will=0),
        attacks=[_melee(id_)],
    )


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


def _engine(
    rng: list[int],
    policy: PolicyFn | None = None,
    llm_policy: LLMPolicy | None = None,
) -> CombatEngine:
    hero = make_state(_combatant("hero"), "players", (0, 0))
    foe = make_state(_combatant("foe"), "monsters", (0, 1))
    states = [hero, foe]
    return CombatEngine(
        ScriptedRng(rng), make_grid(), states,
        policy=policy, llm_policy=llm_policy,
    )


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


def _fixed_policy(*actions: Action) -> PolicyFn:
    def policy(_engine_: CombatEngine, _state: CombatantState) -> tuple[Action, ...]:
        return actions

    return policy


# ── (a) move without a target → missing_target rejection ──────────────────────


def test_move_without_target_is_rejected() -> None:
    eng = _engine([10, 5], policy=_fixed_policy(Action(kind="move")))
    h, _f = eng._states
    eng._roll_initiative()
    eng._take_turn(h)
    assert h.last_action_results == [
        ActionResult(kind="move", ok=False, reason="missing_target")
    ]
    assert any("move rejected (missing_target)" in line for line in eng._transcript)
    # The rejection log line counts as activity: the "wait" fallback must not fire.
    assert not any(line.endswith(": wait") for line in eng._transcript)


# ── (b) declared-but-unimplemented kinds → unsupported_kind ──────────────────


def test_swift_is_rejected_as_unsupported() -> None:
    eng = _engine([10, 5], policy=_fixed_policy(Action(kind="swift")))
    h, _f = eng._states
    eng._roll_initiative()
    eng._take_turn(h)
    assert h.last_action_results == [
        ActionResult(kind="swift", ok=False, reason="unsupported_kind")
    ]
    assert any("swift rejected (unsupported_kind)" in line for line in eng._transcript)


def test_free_and_full_round_are_rejected_as_unsupported() -> None:
    for action in (Action(kind="free"), Action(kind="full_round")):
        eng = _engine([10, 5], policy=_fixed_policy(action))
        h, _f = eng._states
        eng._roll_initiative()
        eng._take_turn(h)
        assert h.last_action_results == [
            ActionResult(kind=action.kind, ok=False, reason="unsupported_kind")
        ]
        expected_log = f"{action.kind} rejected (unsupported_kind)"
        assert any(expected_log in line for line in eng._transcript)


# ── (c) staggered downgrade is recorded AND the executed attack is ok ────────


def test_staggered_full_attack_downgrade_recorded_with_ok_attack() -> None:
    # Rolls: init hero 10, init foe 5, attack d20 10 (10+5 >= AC 10 hits), dmg 3.
    eng = _engine(
        [10, 5, 10, 3],
        policy=_fixed_policy(Action(kind="full_attack", target_id="foe")),
    )
    h, f = eng._states
    _add_condition(h, "staggered")
    eng._roll_initiative()
    eng._take_turn(h)
    results = h.last_action_results
    assert ActionResult(kind="full_attack", ok=False, reason="downgraded:staggered") in results
    assert results[-1] == ActionResult(kind="attack", ok=True, reason="ok")
    assert f.hp == 97


def test_nauseated_full_attack_downgrade_recorded() -> None:
    # Nauseated (B6): only move-like actions survive the filter — full_attack
    # is dropped; only the synthetic downgrade result remains.
    eng = _engine(
        [10, 5],
        policy=_fixed_policy(Action(kind="full_attack", target_id="foe")),
    )
    h, _f = eng._states
    _add_condition(h, "nauseated")
    eng._roll_initiative()
    eng._take_turn(h)
    assert h.last_action_results == [
        ActionResult(kind="full_attack", ok=False, reason="downgraded:nauseated")
    ]


# ── (d) valid attack → ok=True, no new transcript noise ──────────────────────


def test_valid_attack_reports_ok() -> None:
    # Rolls: init hero 10, init foe 5, attack d20 10 (hits AC 10), dmg 3.
    eng = _engine(
        [10, 5, 10, 3],
        policy=_fixed_policy(Action(kind="attack", target_id="foe")),
    )
    h, f = eng._states
    eng._roll_initiative()
    eng._take_turn(h)
    assert h.last_action_results == [
        ActionResult(kind="attack", ok=True, reason="ok")
    ]
    assert f.hp == 97
    assert not any("rejected" in line for line in eng._transcript)


def test_cant_act_clears_results() -> None:
    eng = _engine([10, 5], policy=_fixed_policy(Action(kind="attack", target_id="foe")))
    h, _f = eng._states
    _add_condition(h, "paralyzed")
    h.last_action_results = [ActionResult(kind="attack", ok=True, reason="ok")]
    eng._roll_initiative()
    eng._take_turn(h)
    assert h.last_action_results == []


# ── (e) the Observation exposes last_turn_results with the right shape ───────


def test_snapshot_exposes_last_turn_results() -> None:
    eng = _engine(
        [10, 5, 10, 3],
        policy=_fixed_policy(Action(kind="attack", target_id="foe")),
    )
    h, _f = eng._states
    eng._roll_initiative()
    eng._take_turn(h)
    obs = _snapshot(eng, h)
    hero_c, foe_c = obs["combatants"][0], obs["combatants"][1]
    assert hero_c["last_turn_results"] == (
        {"kind": "attack", "ok": True, "reason": "ok"},
    )
    assert foe_c["last_turn_results"] == ()
    json.dumps(obs)  # JSON-serializable contract


def test_results_visible_in_next_turn_observation() -> None:
    # The Observation is built INSIDE the policy call at the start of a turn:
    # the second invocation must already see the first turn's rejection.
    seen: list[Observation] = []

    def llm_policy(obs: Observation, state: CombatantState) -> tuple[Action, ...]:
        seen.append(obs)
        if len(seen) == 1:
            return (Action(kind="move"),)
        return (Action(kind="wait"),)

    eng = _engine([10, 5], llm_policy=llm_policy)
    h, _f = eng._states
    eng._roll_initiative()
    eng._take_turn(h)
    eng._take_turn(h)
    assert len(seen) == 2
    first_hero = seen[0].combatants[0]
    assert first_hero["last_turn_results"] == ()
    second_hero = seen[1].combatants[0]
    assert second_hero["last_turn_results"] == (
        {"kind": "move", "ok": False, "reason": "missing_target"},
    )