feat(combat): surface explicit action results and reject invalid actions

This commit is contained in:
2026-09-10 11:51:02 +02:00
parent e6125b04f6
commit 4e115f9176
3 changed files with 375 additions and 7 deletions
+11 -1
View File
@@ -297,6 +297,14 @@ Règles modélisées :
suffit à entrer en portée, ou `(move, attack)` sinon. Les actions immédiates 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 (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. 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 ») - Attaques multiples par action : une arme avec `count` > 1 (« 2x Talons »)
résout `count` balayages indépendants dans la même action ; les balayages 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 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 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, l'attaque sournoise), dying (bool — `hp < 0` et pas morte : agonisante,
inconsciente), stabilized (bool — stabilisée pendant son agonie, ne perd 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, - Par paire combattant actif × cible : target_index, distance_ft, weapon_for,
range_penalty, can_charge, can_step5_to_attack, has_line_of_effect, range_penalty, can_charge, can_step5_to_attack, has_line_of_effect,
flanking, movement_cost_ft, target_threatens_me, enemies_threatening_me. flanking, movement_cost_ft, target_threatens_me, enemies_threatening_me.
+115 -6
View File
@@ -340,6 +340,10 @@ class CombatantState:
# Dying & stabilization (not in the frozen data model): the hp at which # Dying & stabilization (not in the frozen data model): the hp at which
# this combatant stabilized while dying; None = not stabilized. # this combatant stabilized while dying; None = not stabilized.
stabilized_hp: int | None = None 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]: def usable_attacks(self) -> list[AttackSpec]:
return [ return [
@@ -516,6 +520,49 @@ class Action:
weapon_id: str | None = None # weapon id for retrieve_weapon / repair_weapon 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) @dataclass(frozen=True)
class ReactionEvent: class ReactionEvent:
"""An off-turn event a ``reaction_policy`` may react to. """An off-turn event a ``reaction_policy`` may react to.
@@ -866,6 +913,10 @@ class CombatEngine:
else 0 else 0
), ),
"engaged_in_melee": self._engaged_in_melee(state), "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( def _pair_snapshot(
@@ -912,27 +963,49 @@ class CombatEngine:
# Check cant_act flag from conditions # Check cant_act flag from conditions
if self.has_condition_flag(state, "cant_act"): if self.has_condition_flag(state, "cant_act"):
self._log(f"round {self._current_round} {state.combatant.id}: wait (cannot 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 return self._winner_side() is not None
self._apply_fear_aura(state) self._apply_fear_aura(state)
state.moved_this_turn = False state.moved_this_turn = False
state.power_attack = False state.power_attack = False
actions = self._policy(self, state) original = self._policy(self, state)
actions = original
results: list[ActionResult] = []
if state.immediate_used: 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 state.immediate_used = False
if self._has_condition_named(state, "staggered"): 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"): 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) before = len(self._transcript)
for action in actions: for action in actions:
if not state.active: if not state.active:
break break
self._execute(state, action) results.append(self._execute(state, action))
if self._winner_side() is not None: if self._winner_side() is not None:
break break
if state.active and len(self._transcript) == before: if state.active and len(self._transcript) == before:
self._log(f"round {self._current_round} {state.combatant.id}: wait") 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 return self._winner_side() is not None
def has_condition_flag(self, state: CombatantState, flag: str) -> bool: def has_condition_flag(self, state: CombatantState, flag: str) -> bool:
@@ -2227,8 +2300,36 @@ class CombatEngine:
attacker, target, weapon, label="charge", bonus_override=bonus 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) 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 kind = action.kind
if kind == "cast_spell": if kind == "cast_spell":
self._execute_cast_spell(state, target, action.spell_name) 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) 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, ...]: def _staggered_filter(actions: tuple[Action, ...]) -> tuple[Action, ...]:
"""Staggered: one move OR one standard action per round, no full-round. """Staggered: one move OR one standard action per round, no full-round.
+249
View File
@@ -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"},
)