diff --git a/tests/test_phase2_measurements.py b/tests/test_phase2_measurements.py new file mode 100644 index 0000000..25e92bb --- /dev/null +++ b/tests/test_phase2_measurements.py @@ -0,0 +1,183 @@ +"""Phase-2 measurement spike (docs/phase-2-llm-architecture.md section 12.b). + +Two numbers decide whether a per-turn LLM is viable for the Monte Carlo +sweep: the OBSERVATION SIZE (bytes / rough tokens) and the POLICY CALLS PER +BATTLE (hence per sweep cell). This module measures both through the real +engine path on a representative caster-heavy encounter, prints a summary +block for the orchestrator, and pins LOOSE upper bounds so a regression +(bigger snapshot schema, policy called twice per turn, ...) fails loudly. +No engine behaviour is changed here: the recorder only reads +``engine._build_observation`` (pure read, consumes no RNG) at each policy +invocation and delegates to ``default_policy``. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import TYPE_CHECKING + +from pf1e_simulator.combat import Action, CombatantState, CombatEngine, 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: + import pytest + + from pf1e_simulator.models import Combatant + +ROOT = Path(__file__).resolve().parents[1] +MAP = ROOT / "data" / "maps" / "sample_arena.yaml" + +# Loose ceilings: a regression past these fails loudly; the real values are +# far below (printed by the tests for the orchestrator to quote). +MAX_OBS_BYTES = 200_000 +MAX_CALLS_PER_BATTLE = 200 + +# Round cap for the measurement battles: generous enough to exercise buffs, +# conditions and summons, small enough to bound the calls/battle figure. +MEASURE_ROUND_CAP = 20 + + +class _Recorder: + """Counting policy wrapper: counts invocations, snapshots observation size. + + Delegates every decision to ``default_policy`` so the battle trajectory + (and therefore the call count) is the real engine path. The observation + is rebuilt at each call via ``engine._build_observation(state)`` - the + same snapshot the engine hands to an ``LLMPolicy`` - and its JSON size + is tracked (first call = start of battle, max = fullest mid-battle + snapshot with buffs/conditions active). + """ + + def __init__(self) -> None: + self.calls = 0 + self.first_bytes: int | None = None + self.max_bytes = 0 + + def policy(self, engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]: + self.calls += 1 + obs = engine._build_observation(state).to_dict() + size = len(json.dumps(obs)) + if self.first_bytes is None: + self.first_bytes = size + self.max_bytes = max(self.max_bytes, size) + 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). + + Oni (channel + heal + summon), misty (spell combat), nairda (hexes) vs + the lich (high-level caster, fear aura, paralysis touch), a thrune + agent, an orc and a goblin: buffs, conditions, spells and pairs all + load the snapshot. + """ + return EncounterSpec( + map=load_map(MAP), + round_cap=MEASURE_ROUND_CAP, + 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 _duel_spec() -> EncounterSpec: + """Small 2v2 baseline to show how calls/battle scales with combatants.""" + return EncounterSpec( + map=load_map(MAP), + round_cap=MEASURE_ROUND_CAP, + sides=( + Side(name="players", combatants=(_c("monsters/goblin.json"),) * 2, zone="A"), + Side(name="monsters", combatants=(_c("monsters/orc.json"),) * 2, zone="B"), + ), + ) + + +def _run_battle( + spec: EncounterSpec, seed: int, registry: dict[str, SpellSpec] +) -> _Recorder: + grid = Grid.from_spec(spec.map) + states = build_states(spec) + rec = _Recorder() + CombatEngine( + SeededRng(seed), + grid, + states, + round_cap=spec.round_cap, + policy=rec.policy, + spell_registry=registry, + ).run() + return rec + + +def _emit(line: str) -> None: + sys.stdout.write(line + "\n") + + +def test_observation_size_and_policy_calls(capsys: pytest.CaptureFixture[str]) -> None: + """Measure both numbers on the representative battle; pin loose bounds.""" + registry = load_spell_registry(SPELLS_DIR) + spec = _representative_spec() + rec = _run_battle(spec, seed=1, registry=registry) + assert rec.first_bytes is not None + + # Determinism: same seed -> same trajectory -> same numbers. + rec2 = _run_battle(spec, seed=1, registry=registry) + assert rec2.calls == rec.calls + assert rec2.first_bytes == rec.first_bytes + assert rec2.max_bytes == rec.max_bytes + + # Loose bounds (regression guards, not exact pins). + assert rec.first_bytes < MAX_OBS_BYTES + assert rec.max_bytes < MAX_OBS_BYTES + assert rec.calls < MAX_CALLS_PER_BATTLE + + duel = _run_battle(_duel_spec(), seed=1, registry=registry) + assert duel.calls < MAX_CALLS_PER_BATTLE + assert duel.first_bytes is not None + + first_tokens = rec.first_bytes // 4 + max_tokens = rec.max_bytes // 4 + _emit("=== PHASE2 SUMMARY (measurement spike) ===") + _emit(f"encounter: 7 combatants, round_cap={MEASURE_ROUND_CAP}, seed=1") + _emit(f"observation bytes (first call): {rec.first_bytes} (~{first_tokens} tokens)") + _emit(f"observation bytes (max in battle): {rec.max_bytes} (~{max_tokens} tokens)") + _emit(f"policy calls/battle (7 combatants): {rec.calls}") + _emit(f"policy calls/battle (2v2 duel): {duel.calls}") + _emit(f"observation 2v2 duel bytes: {duel.first_bytes} (~{duel.first_bytes // 4} tokens)") + for runs in (1, 100, 1000): + _emit(f"projected calls per sweep cell at runs={runs}: {rec.calls * runs}") + out = capsys.readouterr().out + assert "PHASE2 SUMMARY" in out + + +def test_projection_is_linear_in_runs() -> None: + """Calls per sweep cell = calls/battle x runs (fixed per-battle count).""" + registry = load_spell_registry(SPELLS_DIR) + rec = _run_battle(_representative_spec(), seed=7, registry=registry) + assert rec.calls > 0 + assert rec.calls * 1000 == (rec.calls * 100) * 10