feat(phase2): candidate actions

2026-09-21 13:46:05 +02:00
parent 9da9537995
commit 8b5043072a
2 changed files with 582 additions and 0 deletions
@@ -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