diff --git a/src/pf1e_simulator/candidates.py b/src/pf1e_simulator/candidates.py new file mode 100644 index 0000000..ff44b7b --- /dev/null +++ b/src/pf1e_simulator/candidates.py @@ -0,0 +1,384 @@ +"""Plausible candidate actions derived from the observation pairs (P0.3). + +The engine stays the arbiter: ``candidates`` is a pure function of the +``Observation`` snapshot (no ``engine.*`` access) that proposes a +deterministic menu of plausible actions for the acting combatant. Every +candidate is derived from the per-pair decision helpers (``weapon_for``, +``can_charge``, ``can_step5_to_attack``, ``distance_ft``, +``has_line_of_effect``) and the combatant snapshots (``resources_left`` +including ``spell_slot_{level}``, the enriched ``spells`` list). A candidate +that fails a legality gate at execution is rejected with explicit feedback +(``last_turn_results``) — never silently dropped here. + +Candidate categories (docs ``phase-2-llm-architecture.md`` P0.3, D4/D5) +----------------------------------------------------------------------- +- ``attack`` / ``full_attack`` for every enemy the acting combatant can + already hit (``weapon_for`` is set). +- ``charge`` for every enemy whose pair says a straight-line charge is + possible (``can_charge``). +- ``5foot_step`` for every enemy one step brings into reach + (``can_step5_to_attack``). +- ``move`` toward every enemy (auto-pathed by the engine); skipped while + grappled (``grappled_with`` — the engine forbids movement then). +- ``cast_spell`` for every known spell with an available slot + (``spell_slot_{level}``, cantrips free), a castable ``cast_time`` (not + ``long``), at every enemy it reaches (``distance_ft <= range_ft`` and + ``has_line_of_effect`` — the same gate ``_cast_spell`` applies). Harmless + spells (cure, buffs) target the caster: a beneficial spell on a hostile + target would heal or buff the enemy. +- ``channel_energy`` (heal and harm modes) / ``hex`` (evil_eye, slumber per + enemy) / ``deed`` (up_close_and_deadly, quick_clear) when the matching + resource (``channel_energy`` / ``hexes`` / ``grit``) is present. +- ``total_defense`` and ``delay`` (always plausible, no target). +- ``ready`` with each trigger, readied attack on the nearest enemy. + +Kinds emitted (D4): ``attack``, ``full_attack``, ``charge``, ``5foot_step``, +``move``, ``cast_spell``, ``channel_energy``, ``hex``, ``deed``, +``total_defense``, ``ready``, ``delay``. ``swift``/``free``/``full_round`` +are excluded (``full_attack`` is the supported full-round kind). The menu is +capped at ``MAX_CANDIDATES`` (32), trimmed from the tail of the deterministic +order so the core options always survive. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, TypeGuard + +from .combat import Action, Observation + +if TYPE_CHECKING: + from .combat import ReadyTrigger + +# Ready triggers offered for the ``ready`` candidate (the engine's +# ``ReadyTrigger`` literal). +_READY_TRIGGERS: tuple[ReadyTrigger, ...] = ( + "enemy_enters_reach", + "enemy_casts", + "enemy_attacks", + "enemy_hits_ally", +) + +# Class-ability resource keys (in ``resources_left``) that gate the +# channel/hex/deed candidates. +_CHANNEL_RESOURCE = "channel_energy" +_HEX_RESOURCE = "hexes" +_DEED_RESOURCE = "grit" + +# Candidate kinds the generator may emit (D4). The generic minor kinds +# (``swift``/``free``/``full_round``) are excluded: a policy proposes real +# actions, not placeholders. +_CANDIDATE_KINDS: frozenset[str] = frozenset( + { + "attack", + "full_attack", + "charge", + "5foot_step", + "move", + "cast_spell", + "channel_energy", + "hex", + "deed", + "total_defense", + "ready", + "delay", + } +) + +# Deterministic menu order: most decisive first, so the cap trims the least +# useful tail (extra ready triggers, delay) rather than the core options. +_KIND_PRIORITY: dict[str, int] = { + "full_attack": 0, + "attack": 1, + "charge": 2, + "5foot_step": 3, + "move": 4, + "cast_spell": 5, + "channel_energy": 6, + "hex": 7, + "deed": 8, + "total_defense": 9, + "ready": 10, + "delay": 11, +} + +# Upper bound on the returned menu: keeps the LLM prompt compact while +# covering every category. +MAX_CANDIDATES = 32 + + +def candidates(observation: Observation) -> tuple[Action, ...]: + """Deterministic menu of plausible actions for the acting combatant. + + Pure read of the snapshot: no engine access, no RNG, no mutation. The + engine stays the arbiter — a candidate that fails a legality gate is + rejected with explicit feedback, never silently dropped here. + """ + combatants = observation.combatants + if not (0 <= observation.acting_combatant < len(combatants)): + return () + acting = combatants[observation.acting_combatant] + if acting.get("active") is not True: + return () + acting_side = acting.get("side") + pairs = tuple( + pair + for pair in observation.pairs + if _pair_target_is_enemy(pair, combatants, acting_side) + ) + result: list[Action] = [] + _attack_candidates(result, pairs, combatants) + _charge_candidates(result, pairs, combatants) + _step_candidates(result, pairs, combatants) + _move_candidates(result, acting, pairs, combatants) + _spell_candidates(result, acting, pairs, combatants) + _class_ability_candidates(result, acting, pairs, combatants) + result.append(Action(kind="total_defense")) + _ready_candidates(result, pairs, combatants) + result.append(Action(kind="delay")) + result.sort(key=_sort_key) + return tuple(result[:MAX_CANDIDATES]) + + +def _attack_candidates( + result: list[Action], + pairs: tuple[dict[str, object], ...], + combatants: tuple[dict[str, object], ...], +) -> None: + """Standard and full-round attacks for every enemy already in reach.""" + for pair in pairs: + if pair.get("weapon_for") is not None: + target = _target_id(pair, combatants) + result.append(Action(kind="attack", target_id=target)) + result.append(Action(kind="full_attack", target_id=target)) + + +def _charge_candidates( + result: list[Action], + pairs: tuple[dict[str, object], ...], + combatants: tuple[dict[str, object], ...], +) -> None: + """Charge for every enemy with a clear straight-line path.""" + result.extend( + Action(kind="charge", target_id=_target_id(pair, combatants)) + for pair in pairs + if pair.get("can_charge") is True + ) + + +def _step_candidates( + result: list[Action], + pairs: tuple[dict[str, object], ...], + combatants: tuple[dict[str, object], ...], +) -> None: + """5-foot step for every enemy one step brings into reach.""" + result.extend( + Action(kind="5foot_step", target_id=_target_id(pair, combatants)) + for pair in pairs + if pair.get("can_step5_to_attack") is True + ) + + +def _move_candidates( + result: list[Action], + acting: dict[str, object], + pairs: tuple[dict[str, object], ...], + combatants: tuple[dict[str, object], ...], +) -> None: + """Move toward every enemy; skipped while grappled (movement forbidden).""" + if acting.get("grappled_with") is not None: + return + result.extend( + Action(kind="move", target_id=_target_id(pair, combatants)) for pair in pairs + ) + + +def _spell_candidates( + result: list[Action], + acting: dict[str, object], + pairs: tuple[dict[str, object], ...], + combatants: tuple[dict[str, object], ...], +) -> None: + """Cast every known spell with a slot, at every enemy it reaches. + + Harmless spells (cure, buffs) target the caster: a beneficial spell on a + hostile target would heal or buff the enemy. Offensive spells are + proposed per enemy the spell reaches. + """ + spells = acting.get("spells") + resources = acting.get("resources_left") + if not _is_tuple(spells) or not _is_str_object_dict(resources): + return + acting_id = _as_str(acting["id"]) + for spell in spells: + if not _is_str_object_dict(spell): + continue + if not _spell_has_slot(spell, resources): + continue + if _as_str(spell.get("cast_time", "standard")) == "long": + continue + name = _as_str(spell["name"]) + if spell.get("harmless") is True: + result.append( + Action(kind="cast_spell", target_id=acting_id, spell_name=name) + ) + continue + result.extend( + Action( + kind="cast_spell", + target_id=_target_id(pair, combatants), + spell_name=name, + ) + for pair in pairs + if _spell_reaches_pair(spell, pair) + ) + + +def _class_ability_candidates( + result: list[Action], + acting: dict[str, object], + pairs: tuple[dict[str, object], ...], + combatants: tuple[dict[str, object], ...], +) -> None: + """Channel/hex/deed candidates, each gated on its resource being present.""" + resources = acting.get("resources_left") + if not _is_str_object_dict(resources): + return + if _resource_available(resources, _CHANNEL_RESOURCE): + result.append(Action(kind="channel_energy", channel_mode="heal")) + result.append(Action(kind="channel_energy", channel_mode="harm")) + if _resource_available(resources, _HEX_RESOURCE): + for pair in pairs: + target = _target_id(pair, combatants) + result.append(Action(kind="hex", hex_name="evil_eye", target_id=target)) + result.append(Action(kind="hex", hex_name="slumber", target_id=target)) + if _resource_available(resources, _DEED_RESOURCE): + result.append(Action(kind="deed", deed_name="up_close_and_deadly")) + result.append(Action(kind="deed", deed_name="quick_clear")) + + +def _ready_candidates( + result: list[Action], + pairs: tuple[dict[str, object], ...], + combatants: tuple[dict[str, object], ...], +) -> None: + """Ready with each trigger, readied attack on the nearest enemy.""" + nearest = _nearest_enemy(pairs, combatants) + if nearest is None: + return + result.extend( + Action(kind="ready", ready_trigger=trigger, target_id=nearest) + for trigger in _READY_TRIGGERS + ) + + +def _sort_key(action: Action) -> tuple[int, str, str, str, str, str, str]: + """Deterministic ordering of the candidate menu (decisive first).""" + return ( + _KIND_PRIORITY.get(action.kind, 99), + action.target_id or "", + action.spell_name or "", + action.hex_name or "", + action.deed_name or "", + action.ready_trigger or "", + action.channel_mode or "", + ) + + +def _target_id( + pair: dict[str, object], combatants: tuple[dict[str, object], ...] +) -> str: + """Stable id of the pair's target (falls back to the index).""" + index = pair.get("target_index") + if isinstance(index, int) and 0 <= index < len(combatants): + cid = combatants[index].get("id") + if isinstance(cid, str): + return cid + return str(index) + + +def _pair_target_is_enemy( + pair: dict[str, object], + combatants: tuple[dict[str, object], ...], + acting_side: object, +) -> bool: + """True when the pair's target is an active combatant on the other side. + + The observation's pairs cover every other combatant (allies included), + so the side check is what keeps the menu hostile-only. + """ + index = pair.get("target_index") + if not isinstance(index, int) or not (0 <= index < len(combatants)): + return False + target = combatants[index] + return target.get("active") is True and target.get("side") != acting_side + + +def _resource_available(resources: dict[str, object], key: str) -> bool: + """True when the class-ability resource has at least one use left.""" + count = resources.get(key) + return isinstance(count, int) and count > 0 + + +def _spell_has_slot(spell: dict[str, object], resources: dict[str, object]) -> bool: + """True when the caster still has a slot for the spell's level. + + Cantrips (level 0) are cast freely — the snapshot has no level-0 slot + key, mirroring ``_has_slot`` in the engine. + """ + level = spell.get("level") + if not isinstance(level, int): + return False + if level == 0: + return True + count = resources.get(f"spell_slot_{level}") + return isinstance(count, int) and count > 0 + + +def _spell_reaches_pair(spell: dict[str, object], pair: dict[str, object]) -> bool: + """True when the spell's range covers the pair with line of effect. + + Mirrors the engine's ``_spell_reaches`` gate: ``distance_ft`` (the pair's + creature distance) within the snapshot's precomputed ``range_ft``, plus + the pair's ``has_line_of_effect``. + """ + range_ft = spell.get("range_ft") + distance_ft = pair.get("distance_ft") + return ( + isinstance(range_ft, int) + and isinstance(distance_ft, int) + and distance_ft <= range_ft + and pair.get("has_line_of_effect") is True + ) + + +def _nearest_enemy( + pairs: tuple[dict[str, object], ...], combatants: tuple[dict[str, object], ...] +) -> str | None: + """Id of the nearest enemy (lowest ``distance_ft``), or None.""" + best: str | None = None + best_distance: int | None = None + for pair in pairs: + distance = pair.get("distance_ft") + if not isinstance(distance, int): + continue + if best_distance is None or distance < best_distance: + best_distance = distance + best = _target_id(pair, combatants) + return best + + +def _is_tuple(value: object) -> TypeGuard[tuple[object, ...]]: + """Narrow ``object`` to a tuple (snapshot collections are tuples).""" + return isinstance(value, tuple) + + +def _is_str_object_dict(value: object) -> TypeGuard[dict[str, object]]: + """Narrow ``object`` to the snapshot's ``dict[str, object]`` shape.""" + return isinstance(value, dict) + + +def _as_str(value: object) -> str: + if not isinstance(value, str): + msg = f"expected str, got {type(value).__name__}" + raise TypeError(msg) + return value diff --git a/tests/test_phase2_candidates.py b/tests/test_phase2_candidates.py new file mode 100644 index 0000000..07394fb --- /dev/null +++ b/tests/test_phase2_candidates.py @@ -0,0 +1,198 @@ +"""Phase-2 candidate actions (docs/phase-2-llm-architecture.md P0.3). + +``candidates.candidates`` derives a deterministic menu of plausible +``Action`` proposals from the ``Observation`` snapshot's pairs. This module +pins the contract: a normal turn yields a non-empty, capped menu whose kinds +stay within the declared set; the menu is deterministic; no candidate ever +targets an ally (the pairs cover every other combatant, so the side filter is +what keeps the menu hostile-only); a weaponless combatant still gets +movement/defensive options; an enemy in weapon reach gets attack/full_attack; +and harmless spells (cure, buffs) target the caster rather than an enemy. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from pf1e_simulator.candidates import _CANDIDATE_KINDS, MAX_CANDIDATES, candidates +from pf1e_simulator.combat import ( + Action, + CombatantState, + CombatEngine, + Observation, + default_policy, +) +from pf1e_simulator.grid import Grid +from pf1e_simulator.loaders import load_combatant +from pf1e_simulator.map import load_map +from pf1e_simulator.rng import SeededRng +from pf1e_simulator.runner import EncounterSpec, Side, build_states +from pf1e_simulator.spells import SPELLS_DIR, SpellSpec, load_spell_registry + +if TYPE_CHECKING: + from pf1e_simulator.models import Combatant + +ROOT = Path(__file__).resolve().parents[1] +MAP = ROOT / "data" / "maps" / "sample_arena.yaml" + + +class _CaptureAll: + """Policy wrapper that snapshots every observation, then delegates.""" + + def __init__(self) -> None: + self.observations: dict[str, list[Observation]] = {} + + def policy(self, engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]: + obs = engine._build_observation(state) + self.observations.setdefault(state.combatant.id, []).append(obs) + return default_policy(engine, state) + + +def _c(path: str) -> Combatant: + return load_combatant(ROOT / "data" / path) + + +def _representative_spec() -> EncounterSpec: + """Caster-heavy 3v4 on the sample arena (walls, rubble, water, pillars).""" + return EncounterSpec( + map=load_map(MAP), + round_cap=20, + sides=( + Side( + name="players", + combatants=(_c("pcs/oni.json"), _c("pcs/misty.json"), _c("pcs/nairda.json")), + zone="A", + ), + Side( + name="monsters", + combatants=( + _c("monsters/lich.json"), + _c("monsters/thrune_agent.json"), + _c("monsters/orc.json"), + _c("monsters/goblin.json"), + ), + zone="B", + ), + ), + ) + + +def _weaponless_spec() -> EncounterSpec: + """1v1: a goblin with no usable weapons against an orc.""" + goblin = _c("monsters/goblin.json").model_copy(update={"attacks": ()}) + return EncounterSpec( + map=load_map(MAP), + round_cap=20, + sides=( + Side(name="players", combatants=(goblin,), zone="A"), + Side(name="monsters", combatants=(_c("monsters/orc.json"),), zone="B"), + ), + ) + + +def _run_observations( + spec: EncounterSpec, seed: int, registry: dict[str, SpellSpec] +) -> dict[str, list[Observation]]: + grid = Grid.from_spec(spec.map) + states = build_states(spec) + cap = _CaptureAll() + CombatEngine( + SeededRng(seed), + grid, + states, + round_cap=spec.round_cap, + policy=cap.policy, + spell_registry=registry, + ).run() + return cap.observations + + +def _pair_target_side(pair: dict[str, object], obs: Observation) -> object: + """Side of the pair's target combatant, or None when out of range.""" + index = pair.get("target_index") + if not isinstance(index, int) or not (0 <= index < len(obs.combatants)): + return None + return obs.combatants[index].get("side") + + +def test_candidates_normal_turn_non_empty_and_supported_kinds() -> None: + registry = load_spell_registry(SPELLS_DIR) + observations = _run_observations(_representative_spec(), 1, registry) + obs = next(obs_list[0] for obs_list in observations.values()) + menu = candidates(obs) + assert menu + assert len(menu) <= MAX_CANDIDATES + kinds = {a.kind for a in menu} + assert kinds <= _CANDIDATE_KINDS + assert {"move", "total_defense", "delay"} <= kinds + + +def test_candidates_deterministic() -> None: + registry = load_spell_registry(SPELLS_DIR) + observations = _run_observations(_representative_spec(), 1, registry) + obs = next(obs_list[0] for obs_list in observations.values()) + assert candidates(obs) == candidates(obs) + + +def test_candidates_no_allies_targeted() -> None: + registry = load_spell_registry(SPELLS_DIR) + observations = _run_observations(_representative_spec(), 1, registry) + for obs_list in observations.values(): + for obs in obs_list: + acting = obs.combatants[obs.acting_combatant] + acting_side = acting["side"] + acting_id = acting["id"] + for action in candidates(obs): + if action.target_id is None or action.target_id == acting_id: + continue + target = next( + (c for c in obs.combatants if c["id"] == action.target_id), None + ) + assert target is not None + assert target["side"] != acting_side + + +def test_candidates_weaponless_gets_movement_and_defense() -> None: + registry = load_spell_registry(SPELLS_DIR) + observations = _run_observations(_weaponless_spec(), 1, registry) + obs = observations["goblin"][0] + kinds = {a.kind for a in candidates(obs)} + assert {"move", "total_defense", "ready", "delay"} <= kinds + assert not ({"attack", "full_attack", "charge"} & kinds) + + +def test_candidates_in_range_gets_attack() -> None: + registry = load_spell_registry(SPELLS_DIR) + observations = _run_observations(_representative_spec(), 1, registry) + for obs_list in observations.values(): + for obs in obs_list: + acting = obs.combatants[obs.acting_combatant] + acting_side = acting["side"] + enemy_pairs = [ + p for p in obs.pairs if _pair_target_side(p, obs) != acting_side + ] + if not any(p.get("weapon_for") is not None for p in enemy_pairs): + continue + kinds = {a.kind for a in candidates(obs)} + assert "attack" in kinds + assert "full_attack" in kinds + return + msg = "no observation with an enemy in weapon reach found" + raise AssertionError(msg) + + +def test_candidates_harmless_spells_target_self() -> None: + registry = load_spell_registry(SPELLS_DIR) + observations = _run_observations(_representative_spec(), 1, registry) + obs = observations["oni"][0] + acting = obs.combatants[obs.acting_combatant] + acting_id = acting["id"] + harmless_casts = [ + a + for a in candidates(obs) + if a.kind == "cast_spell" + and a.spell_name in {"Guidance", "Shield of Faith", "Stabilize"} + ] + assert harmless_casts + assert all(a.target_id == acting_id for a in harmless_casts)