pf1e-simulator/tests/test_phase2_view.py

"""Phase-2 compact observation view (docs/phase-2-llm-architecture.md P0.2).

``llm_view.build_view`` + ``render`` turn the raw ``Observation`` snapshot
into a compact, deterministic text view for the LLM prompt. This module
measures the compaction on a representative caster-heavy encounter (same
shape as the measurement spike, but with its OWN spec builder - the spike
file is out of scope), prints a summary block for the orchestrator, and
pins LOOSE upper bounds so a regression (a dropped key re-added, a sparse
grid turned dense, ...) fails loudly. It also pins the structural contract:
sparse grids only carry non-default cells, dropped keys stay dropped, the
nested attack/spell expansions stay compact, and the render is
deterministic.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import TYPE_CHECKING

from pf1e_simulator.combat import (
    OBSERVATION_SCHEMA_VERSION,
    Action,
    CombatantState,
    CombatEngine,
    Observation,
    default_policy,
)
from pf1e_simulator.grid import Grid
from pf1e_simulator.llm_view import build_view, render
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 (regression guards, not exact pins): the docs target a
# compact view of ~2-4k tokens against a raw snapshot of ~16k tokens, i.e.
# 12.5-25% of the raw size. 25% of raw bytes and a 4k-token byte ceiling
# (4 bytes/token) are the outer bounds.
MAX_COMPACT_RATIO = 0.25
MAX_COMPACT_BYTES = 16_000

# Keys the compact view deliberately drops from each combatant snapshot.
_DROPPED_COMBATANT_KEYS = frozenset(
    {
        "occupied_cells",
        "abilities",
        "skills",
        "initiative_mod",
        "initiative_total",
        "temp_hp_rounds_left",
        "features",
        "feature_details",
        "carried_items",
        "weapon_damage",
        "aware",
        "alignment",
        "armor_spell_failure",
        "arcane_caster",
        "run_rounds",
    }
)

# Every pair key must survive (the pairs are the tactical core); the raw
# ``target_index`` is replaced by the stable combatant ``target`` id.
_PAIR_KEYS = frozenset(
    {
        "target",
        "distance_ft",
        "weapon_for",
        "weapon_reason",
        "range_penalty",
        "can_charge",
        "charge_reason",
        "can_step5_to_attack",
        "has_line_of_effect",
        "flanking",
        "movement_cost_ft",
        "target_threatens_me",
        "enemies_threatening_me",
        "cover_bonus",
    }
)

# Keys dropped from the nested attack/spell expansions.
_ATTACK_DROPPED_KEYS = frozenset({"splash", "hardness", "hit_points"})
_SPELL_DROPPED_KEYS = frozenset(
    {"summary", "school", "half_on_save", "save_negates", "harmless", "components"}
)


class _Capture:
    """Policy wrapper that snapshots the first observation, then delegates."""

    def __init__(self) -> None:
        self.observation: Observation | None = None

    def policy(self, engine: CombatEngine, state: CombatantState) -> tuple[Action, ...]:
        if self.observation is None:
            self.observation = engine._build_observation(state)
        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 _first_observation(
    spec: EncounterSpec, seed: int, registry: dict[str, SpellSpec]
) -> Observation:
    grid = Grid.from_spec(spec.map)
    states = build_states(spec)
    cap = _Capture()
    CombatEngine(
        SeededRng(seed),
        grid,
        states,
        round_cap=spec.round_cap,
        policy=cap.policy,
        spell_registry=registry,
    ).run()
    assert cap.observation is not None
    return cap.observation


def _emit(line: str) -> None:
    sys.stdout.write(line + "\n")


def _is_default_terrain(semantics: dict[str, object]) -> bool:
    """Mirror of ``llm_view._is_default_terrain`` for the structural test."""
    return (
        semantics.get("move_cost") == 1
        and not semantics.get("blocks_los")
        and not semantics.get("cover")
        and semantics.get("cover_bonus") == 0
    )


def test_compact_view_size(capsys: pytest.CaptureFixture[str]) -> None:
    """Measure raw-vs-compact on the representative battle; pin loose bounds."""
    registry = load_spell_registry(SPELLS_DIR)
    obs = _first_observation(_representative_spec(), seed=1, registry=registry)
    raw = json.dumps(obs.to_dict())
    view = build_view(obs)
    compact = render(view)

    raw_bytes = len(raw)
    compact_bytes = len(compact)

    # Determinism: rebuilding the view from the same observation renders
    # byte-identically.
    assert render(build_view(obs)) == compact

    # Schema version is pinned to the engine's contract.
    assert view.schema_version == OBSERVATION_SCHEMA_VERSION

    # Loose bounds (regression guards, not exact pins).
    assert compact_bytes < raw_bytes * MAX_COMPACT_RATIO
    assert compact_bytes < MAX_COMPACT_BYTES

    _emit("=== PHASE2 SUMMARY (compact observation view) ===")
    _emit("encounter: 7 combatants, round_cap=20, seed=1")
    _emit(f"raw observation bytes:      {raw_bytes}  (~{raw_bytes // 4} tokens)")
    _emit(f"compact view bytes:         {compact_bytes}  (~{compact_bytes // 4} tokens)")
    _emit(f"compaction ratio:           {compact_bytes / raw_bytes:.1%} of raw")
    _emit(f"combatants in view:         {len(view.combatants)}")
    _emit(f"pairs in view:              {len(view.pairs)}")
    _emit(f"sparse terrain cells:       {len(view.terrain)}")
    _emit(f"sparse elevation cells:     {len(view.elevation)}")
    _emit(f"threatened cells:           {len(view.threatened)}")
    _emit(f"ranged_threatened cells:    {len(view.ranged_threatened)}")
    out = capsys.readouterr().out
    assert "PHASE2 SUMMARY" in out


def test_view_sparse_grids() -> None:
    """Pin the sparse-grid contract: only non-default/non-zero/True cells."""
    registry = load_spell_registry(SPELLS_DIR)
    obs = _first_observation(_representative_spec(), seed=1, registry=registry)
    raw = obs.to_dict()
    view = build_view(obs)
    semantics_by_type = {str(sem["type"]): sem for sem in raw["terrain_semantics"]}

    # Dimensions match the raw terrain grid.
    assert view.dimensions == (len(raw["terrain"]), len(raw["terrain"][0]))

    # Sparse terrain only carries non-default cells, with the raw char.
    for r, c, terrain_char in view.terrain:
        assert not _is_default_terrain(semantics_by_type[terrain_char])
        assert raw["terrain"][r][c] == terrain_char

    # Sparse elevation only carries non-zero cells.
    for r, c, height in view.elevation:
        assert height != 0
        assert raw["elevation"][r][c] == height

    # Sparse threatened grids only carry True cells.
    for r, c in view.threatened:
        assert raw["threatened"][r][c] is True
    for r, c in view.ranged_threatened:
        assert raw["ranged_threatened"][r][c] is True


def test_view_dropped_keys() -> None:
    """Pin the key contract: dropped keys stay dropped, pair keys survive."""
    registry = load_spell_registry(SPELLS_DIR)
    obs = _first_observation(_representative_spec(), seed=1, registry=registry)
    view = build_view(obs)

    for combatant in view.combatants:
        assert _DROPPED_COMBATANT_KEYS.isdisjoint(combatant.keys())
    for pair in view.pairs:
        assert _PAIR_KEYS.issubset(pair.keys())

    # Nested expansions stay compact: no dropped attack/spell keys, and the
    # attack damage is rendered as formula+type strings.
    for combatant in view.combatants:
        for attack in combatant["attacks"]:
            assert _ATTACK_DROPPED_KEYS.isdisjoint(attack.keys())
            assert all(isinstance(d, str) for d in attack["damage"])
        for spell in combatant["spells"]:
            assert _SPELL_DROPPED_KEYS.isdisjoint(spell.keys())


def test_view_ids_and_render() -> None:
    """Pin the id contract and the deterministic line-based render."""
    registry = load_spell_registry(SPELLS_DIR)
    obs = _first_observation(_representative_spec(), seed=1, registry=registry)
    raw = obs.to_dict()
    view = build_view(obs)

    # The acting combatant and pair targets are rendered as ids.
    assert view.acting_combatant in raw["initiative_order"]
    ids = {combatant["id"] for combatant in view.combatants}
    for pair in view.pairs:
        assert pair["target"] in ids

    # The render is line-based and self-describing.
    text = render(view)
    assert text.startswith("schema_version=1\n")
    assert "combatants:" in text
    assert "pairs:" in text
    assert "condition_definitions" not in text