test(phase2): measurement spike - observation size and policy calls per battle

Representative 7-combatant caster-heavy encounter on sample_arena, measured through the real engine path (engine._build_observation at each policy call, delegating to default_policy): - observation: 64,204 bytes at first call, 67,712 max mid-battle (~16.9k tokens) - policy calls/battle: 124 (7 combatants, round_cap=20), 12 for a 2v2 duel - projected calls per sweep cell: 124 / 12,400 / 124,000 at runs 1/100/1000 Loose bounds (obs < 200 KB, calls/battle < 200) + seed-determinism pinned.
2026-09-21 11:26:24 +02:00
parent 3c3e30634e
commit e5c4892a57
1 changed files with 183 additions and 0 deletions
@@ -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