ctan/pf1e-simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(phase2): candidate actions

Browse Source
This commit is contained in:
Thien An
2026-09-21 13:46:05 +02:00
parent 9da9537995
commit 8b5043072a
2 changed files with 582 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pf1e_simulator/candidates.py
+384
View File
@@ -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
+198
View File
@@ -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)