154 lines
5.0 KiB
Python
154 lines
5.0 KiB
Python
"""Phase-2 runner policy entry points (docs/phase-2-llm-architecture.md P0.4).
|
|
|
|
``run_encounter`` grows two Phase-2 keyword arguments: ``llm_policy`` (the
|
|
observation-only/legacy policy every ``CombatEngine`` gets) and
|
|
``policy_factory`` (a per-run ``Policy`` builder called once per battle with
|
|
the run index). These tests pin the contract: the factory is invoked exactly
|
|
once per run in index order, its result is the run's policy (``None`` falls
|
|
back to ``default_policy``), the three entry points are mutually exclusive,
|
|
and an ``llm_policy`` really drives a full battle through the runner.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
|
|
from pf1e_simulator.combat import Action, default_policy
|
|
from pf1e_simulator.loaders import load_combatant
|
|
from pf1e_simulator.map import load_map
|
|
from pf1e_simulator.runner import EncounterSpec, Side, run_encounter
|
|
from pf1e_simulator.spells import SPELLS_DIR, load_spell_registry
|
|
|
|
if TYPE_CHECKING:
|
|
from pf1e_simulator.combat import Observation, Policy
|
|
from pf1e_simulator.models import Combatant
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MAP = ROOT / "data" / "maps" / "sample_arena.yaml"
|
|
|
|
|
|
def _c(path: str) -> Combatant:
|
|
return load_combatant(ROOT / "data" / path)
|
|
|
|
|
|
def _duel_spec() -> EncounterSpec:
|
|
"""A fast 1v1 (goblin archer vs orc) on the sample arena."""
|
|
return EncounterSpec(
|
|
map=load_map(MAP),
|
|
round_cap=20,
|
|
sides=(
|
|
Side(name="players", combatants=(_c("monsters/goblin.json"),), zone="A"),
|
|
Side(name="monsters", combatants=(_c("monsters/orc.json"),), zone="B"),
|
|
),
|
|
)
|
|
|
|
|
|
def _wait_policy(_observation: Observation) -> tuple[Action, ...]:
|
|
"""An observation-only policy that never attacks (forces a draw)."""
|
|
return (Action(kind="wait"),)
|
|
|
|
|
|
def test_policy_factory_called_once_per_run_in_order() -> None:
|
|
"""The factory sees each run index once, and its policy drives the run."""
|
|
spec = _duel_spec()
|
|
registry = load_spell_registry(SPELLS_DIR)
|
|
calls: list[int] = []
|
|
|
|
def factory(index: int) -> Policy | None:
|
|
calls.append(index)
|
|
return default_policy
|
|
|
|
report = run_encounter(
|
|
spec, runs=4, seed=3, policy_factory=factory, spell_registry=registry
|
|
)
|
|
direct = run_encounter(
|
|
spec, runs=4, seed=3, policy=default_policy, spell_registry=registry
|
|
)
|
|
|
|
assert calls == [0, 1, 2, 3]
|
|
# A per-run factory returning default_policy is bit-identical to passing
|
|
# default_policy directly (same seed stream, same policy each run).
|
|
assert report.wins == direct.wins
|
|
assert report.draws == direct.draws
|
|
assert report.avg_rounds == direct.avg_rounds
|
|
assert report.attrition == direct.attrition
|
|
|
|
|
|
def test_policy_factory_none_falls_back_to_default_policy() -> None:
|
|
"""A factory returning None leaves the engine on default_policy."""
|
|
spec = _duel_spec()
|
|
registry = load_spell_registry(SPELLS_DIR)
|
|
|
|
report = run_encounter(
|
|
spec,
|
|
runs=2,
|
|
seed=5,
|
|
policy_factory=lambda _index: None,
|
|
spell_registry=registry,
|
|
)
|
|
direct = run_encounter(spec, runs=2, seed=5, spell_registry=registry)
|
|
|
|
assert report.wins == direct.wins
|
|
assert report.avg_rounds == direct.avg_rounds
|
|
|
|
|
|
def test_policy_entry_points_mutually_exclusive() -> None:
|
|
"""policy / llm_policy / policy_factory reject every conflicting pairing."""
|
|
spec = _duel_spec()
|
|
registry = load_spell_registry(SPELLS_DIR)
|
|
|
|
def factory(_index: int) -> Policy | None:
|
|
return default_policy
|
|
|
|
with pytest.raises(ValueError, match="policy_factory and policy"):
|
|
run_encounter(
|
|
spec,
|
|
runs=1,
|
|
seed=1,
|
|
policy=default_policy,
|
|
policy_factory=factory,
|
|
spell_registry=registry,
|
|
)
|
|
with pytest.raises(ValueError, match="policy_factory and llm_policy"):
|
|
run_encounter(
|
|
spec,
|
|
runs=1,
|
|
seed=1,
|
|
llm_policy=_wait_policy,
|
|
policy_factory=factory,
|
|
spell_registry=registry,
|
|
)
|
|
with pytest.raises(ValueError, match="policy and llm_policy"):
|
|
run_encounter(
|
|
spec,
|
|
runs=1,
|
|
seed=1,
|
|
policy=default_policy,
|
|
llm_policy=_wait_policy,
|
|
spell_registry=registry,
|
|
)
|
|
|
|
|
|
def test_llm_policy_drives_a_full_battle_through_the_runner() -> None:
|
|
"""The observation-only policy is invoked each turn and settles the battle."""
|
|
spec = _duel_spec()
|
|
registry = load_spell_registry(SPELLS_DIR)
|
|
seen: list[int] = []
|
|
|
|
def wait(observation: Observation) -> tuple[Action, ...]:
|
|
seen.append(observation.round)
|
|
return (Action(kind="wait"),)
|
|
|
|
report = run_encounter(
|
|
spec, runs=1, seed=1, llm_policy=wait, spell_registry=registry
|
|
)
|
|
|
|
# Both combatants wait every round: the policy is called for every turn
|
|
# and nobody can win, so the battle draws at the round cap.
|
|
assert seen
|
|
assert report.draws == 1
|
|
assert report.wins == {"players": 0, "monsters": 0}
|