diff --git a/README.md b/README.md index ecf391e..30b65a2 100644 --- a/README.md +++ b/README.md @@ -297,6 +297,14 @@ Règles modélisées : suffit à entrer en portée, ou `(move, attack)` sinon. Les actions immédiates (hors-tour, consomment le prochain swift) sont modélisées via une politique de réaction opt-in (`reaction_policy`) — voir « Actions immédiates » ci-dessous. +- Retour d'exécution des actions : chaque action produit un `ActionResult` + explicite (kind, ok, reason) consigné dans le snapshot (`last_turn_results` + au tour suivant). Les actions invalides sont rejetées et journalisées au lieu + d'être des no-op silencieux : cible absente → `missing_target` ; kinds + déclarés mais non implémentés (`swift`, `free`, `full_round` — leur + implémentation générique reste à faire) → `unsupported_kind` ; dégradation + par les filtres staggered/nauseated (ou swift consommé par une immédiate) → + `downgraded:staggered` / `downgraded:nauseated` / `downgraded:immediate_swift`. - Attaques multiples par action : une arme avec `count` > 1 (« 2x Talons ») résout `count` balayages indépendants dans la même action ; les balayages restants sont perdus si la cible tombe (inconsciente ou morte) en cours de @@ -547,7 +555,9 @@ Règles modélisées : une cible engagée dans une mêlée, signal pour le tir dans la mêlée et l'attaque sournoise), dying (bool — `hp < 0` et pas morte : agonisante, inconsciente), stabilized (bool — stabilisée pendant son agonie, ne perd - plus de pv tant qu'aucune nouvelle blessure ne la remet en agonie). + plus de pv tant qu'aucune nouvelle blessure ne la remet en agonie), + last_turn_results (tuple de {kind, ok, reason} — résultat d'exécution du + dernier tour, pour que la policy apprenne les actions rejetées/dégradées). - Par paire combattant actif × cible : target_index, distance_ft, weapon_for, range_penalty, can_charge, can_step5_to_attack, has_line_of_effect, flanking, movement_cost_ft, target_threatens_me, enemies_threatening_me. diff --git a/src/pf1e_simulator/combat.py b/src/pf1e_simulator/combat.py index 19608ac..fda0bc5 100644 --- a/src/pf1e_simulator/combat.py +++ b/src/pf1e_simulator/combat.py @@ -340,6 +340,10 @@ class CombatantState: # Dying & stabilization (not in the frozen data model): the hp at which # this combatant stabilized while dying; None = not stabilized. stabilized_hp: int | None = None + # Action feedback (not in the frozen data model): the ActionResults of the + # PREVIOUS turn, assigned at the end of each turn so the Observation built + # at the next policy invocation can expose them. + last_action_results: list[ActionResult] = field(default_factory=list) def usable_attacks(self) -> list[AttackSpec]: return [ @@ -516,6 +520,49 @@ class Action: weapon_id: str | None = None # weapon id for retrieve_weapon / repair_weapon +# Normalized action-result reason vocabulary. U1 emits ok, unsupported_kind, +# missing_target and the downgraded:* reasons; no_weapon, out_of_range, +# no_line_of_effect, no_slot and insufficient_resource are declared here for +# the deeper execution-feedback package (U6b). +ACTION_RESULT_REASONS: frozenset[str] = frozenset({ + "ok", "unsupported_kind", "missing_target", "no_weapon", "out_of_range", + "no_line_of_effect", "no_slot", "insufficient_resource", + "downgraded:staggered", "downgraded:nauseated", +}) +_REASON_OK = "ok" +_REASON_UNSUPPORTED_KIND = "unsupported_kind" +_REASON_MISSING_TARGET = "missing_target" +_REASON_DOWNGRADED_STAGGERED = "downgraded:staggered" +_REASON_DOWNGRADED_NAUSEATED = "downgraded:nauseated" +_REASON_DOWNGRADED_IMMEDIATE_SWIFT = "downgraded:immediate_swift" + +# Action kinds that are declared but have no engine dispatch (U1 surfaces +# them as explicit rejections instead of silent no-ops). +_UNSUPPORTED_ACTION_KINDS: frozenset[str] = frozenset({"swift", "free", "full_round"}) +# Action kinds whose helper is a silent no-op (or worse) without a resolved +# target; _execute rejects them up front with missing_target instead. +_TARGET_REQUIRED_ACTION_KINDS: frozenset[str] = frozenset({ + "attack", "full_attack", "charge", "withdraw", "move", "5foot_step", + "cast_spell", "maneuver", +}) + + +@dataclass(frozen=True) +class ActionResult: + """Explicit outcome of executing one Action (feedback for the policy). + + ``ok`` is True only when the action was dispatched to its helper; + rejections (``unsupported_kind``, ``missing_target``) and filter + downgrades (``downgraded:*``) carry ok=False and a reason drawn from + ``ACTION_RESULT_REASONS``. ``detail`` is reserved for future packages. + """ + + kind: str + ok: bool + reason: str = "" + detail: str = "" + + @dataclass(frozen=True) class ReactionEvent: """An off-turn event a ``reaction_policy`` may react to. @@ -866,6 +913,10 @@ class CombatEngine: else 0 ), "engaged_in_melee": self._engaged_in_melee(state), + "last_turn_results": tuple( + {"kind": r.kind, "ok": r.ok, "reason": r.reason} + for r in state.last_action_results + ), } def _pair_snapshot( @@ -912,27 +963,49 @@ class CombatEngine: # Check cant_act flag from conditions if self.has_condition_flag(state, "cant_act"): self._log(f"round {self._current_round} {state.combatant.id}: wait (cannot act)") + state.last_action_results = [] return self._winner_side() is not None self._apply_fear_aura(state) state.moved_this_turn = False state.power_attack = False - actions = self._policy(self, state) + original = self._policy(self, state) + actions = original + results: list[ActionResult] = [] if state.immediate_used: - actions = _without_swift_bonus(actions) + kept = _without_swift_bonus(actions) + results.extend( + ActionResult(kind=kind, ok=False, reason=_REASON_DOWNGRADED_IMMEDIATE_SWIFT) + for kind in _dropped_kinds(actions, kept) + ) + actions = kept state.immediate_used = False if self._has_condition_named(state, "staggered"): - actions = _staggered_filter(actions) + kept = _staggered_filter(actions) + results.extend( + ActionResult(kind=kind, ok=False, reason=_REASON_DOWNGRADED_STAGGERED) + for kind in _dropped_kinds(actions, kept) + ) + actions = kept elif self.has_condition_flag(state, "cant_attack"): - actions = _nauseated_filter(actions) + kept = _nauseated_filter(actions) + results.extend( + ActionResult(kind=kind, ok=False, reason=_REASON_DOWNGRADED_NAUSEATED) + for kind in _dropped_kinds(actions, kept) + ) + actions = kept before = len(self._transcript) for action in actions: if not state.active: break - self._execute(state, action) + results.append(self._execute(state, action)) if self._winner_side() is not None: break if state.active and len(self._transcript) == before: self._log(f"round {self._current_round} {state.combatant.id}: wait") + # Assigned at the END of the turn (never reset at the start): the + # Observation is built inside the policy call at the START of the next + # turn, so it must still see THIS turn's results at that moment. + state.last_action_results = results return self._winner_side() is not None def has_condition_flag(self, state: CombatantState, flag: str) -> bool: @@ -2227,8 +2300,36 @@ class CombatEngine: attacker, target, weapon, label="charge", bonus_override=bonus ) - def _execute(self, state: CombatantState, action: Action) -> None: + def _execute(self, state: CombatantState, action: Action) -> ActionResult: + """Execute one action and return its explicit result. + + Rejections (unsupported kinds, missing target) are logged and never + reach the helpers; everything else dispatches exactly as before and + reports ok. + """ + kind = action.kind + if kind in _UNSUPPORTED_ACTION_KINDS: + self._log( + f"round {self._current_round} {state.combatant.id}: " + f"{kind} rejected ({_REASON_UNSUPPORTED_KIND})" + ) + return ActionResult( + kind=kind, ok=False, reason=_REASON_UNSUPPORTED_KIND + ) target = next((s for s in self._states if s.combatant.id == action.target_id), None) + if target is None and kind in _TARGET_REQUIRED_ACTION_KINDS: + self._log( + f"round {self._current_round} {state.combatant.id}: " + f"{kind} rejected ({_REASON_MISSING_TARGET})" + ) + return ActionResult(kind=kind, ok=False, reason=_REASON_MISSING_TARGET) + self._dispatch_action(state, target, action) + return ActionResult(kind=kind, ok=True, reason=_REASON_OK) + + def _dispatch_action( + self, state: CombatantState, target: CombatantState | None, action: Action + ) -> None: + """Route a validated action to its execution helper.""" kind = action.kind if kind == "cast_spell": self._execute_cast_spell(state, target, action.spell_name) @@ -4016,6 +4117,14 @@ def _without_swift_bonus(actions: tuple[Action, ...]) -> tuple[Action, ...]: return tuple(a for a in actions if a.kind not in _SWIFT_BONUS_KINDS) +def _dropped_kinds( + before: tuple[Action, ...], after: tuple[Action, ...] +) -> tuple[str, ...]: + """Kinds present in ``before`` but absent from the filtered ``after``.""" + kept = {a.kind for a in after} + return tuple(a.kind for a in before if a.kind not in kept) + + def _staggered_filter(actions: tuple[Action, ...]) -> tuple[Action, ...]: """Staggered: one move OR one standard action per round, no full-round. diff --git a/tests/test_action_feedback.py b/tests/test_action_feedback.py new file mode 100644 index 0000000..cc7d58a --- /dev/null +++ b/tests/test_action_feedback.py @@ -0,0 +1,249 @@ +"""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: full_attack is not a standard action → the filter empties the + # sequence; 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"}, + )