feat(phase2): candidate actions

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

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)