pf1e-simulator/tests/test_phase2_sweep.py

"""Phase-2 sweep runner tests (docs/phase-2-llm-architecture.md 10 P2.3, D9/D10).

The ``SweepRunner`` executes a grid of (scenario x strategy) cells in
parallel, each cell running ``run_encounter(..., collect_runs=True)`` under
one policy and reporting the aggregate report plus the party's danger
metrics. These tests pin the contract: scenario-major grid order, bit-for-bit
determinism, thread-pool equivalence (workers=1 vs workers=4), the
``max_total_runs`` cost bound, constructor/cell validation, the equivalence
of a ``None`` policy cell with a direct ``run_encounter`` call, and the
aligned text renderer. All cells use small n (runs <= 5) so the file stays
fast and deterministic.
"""

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING

import pytest

from pf1e_simulator.loaders import load_combatant
from pf1e_simulator.map import load_map
from pf1e_simulator.objectives import danger_metrics
from pf1e_simulator.phase2_spike import focus_fire_policy
from pf1e_simulator.runner import EncounterSpec, Side, run_encounter
from pf1e_simulator.sweep import SweepCell, SweepRunner, format_sweep

if TYPE_CHECKING:
    from pf1e_simulator.models import Combatant

ROOT = Path(__file__).resolve().parents[1]
MAP = ROOT / "data" / "maps" / "sample_arena.yaml"

RUNS = 5
SEED = 7
ROUND_CAP = 20
PARTY = "players"


def _c(path: str) -> Combatant:
    return load_combatant(ROOT / "data" / path)


def _spec(goblins: int, orcs: int) -> EncounterSpec:
    return EncounterSpec(
        map=load_map(MAP),
        round_cap=ROUND_CAP,
        sides=(
            Side(
                name="players",
                combatants=tuple(_c("monsters/goblin.json") for _ in range(goblins)),
                zone="A",
            ),
            Side(
                name="monsters",
                combatants=tuple(_c("monsters/orc.json") for _ in range(orcs)),
                zone="B",
            ),
        ),
    )


VARIANT_A = _spec(goblins=3, orcs=2)
VARIANT_B = _spec(goblins=2, orcs=3)

SCENARIOS = (("variant_a", VARIANT_A), ("variant_b", VARIANT_B))
STRATEGIES = (("default", None), ("focus_fire_monsters", focus_fire_policy("monsters")))


def _cells() -> tuple[SweepCell, ...]:
    return (
        SweepCell(
            scenario="variant_a",
            strategy="default",
            spec=VARIANT_A,
            policy=None,
            runs=RUNS,
            seed=SEED,
        ),
        SweepCell(
            scenario="variant_b",
            strategy="focus_fire_monsters",
            spec=VARIANT_B,
            policy=focus_fire_policy("monsters"),
            runs=RUNS,
            seed=SEED,
        ),
    )


def test_run_grid_2x2_scenario_major() -> None:
    """A 2x2 grid returns 4 results in scenario-major order with sane metrics."""
    runner = SweepRunner(party=PARTY, workers=2)
    results = runner.run_grid(SCENARIOS, STRATEGIES, runs=RUNS, seed=SEED)

    assert len(results) == 4
    assert [r.cell.scenario for r in results] == [
        "variant_a",
        "variant_a",
        "variant_b",
        "variant_b",
    ]
    assert [r.cell.strategy for r in results] == [
        "default",
        "focus_fire_monsters",
        "default",
        "focus_fire_monsters",
    ]
    for result in results:
        assert result.report.runs == RUNS
        assert result.danger.runs == RUNS
        assert 0.0 <= result.danger.p_survives <= 1.0
        assert 0.0 <= result.danger.p_tpk <= 1.0


def test_run_is_deterministic() -> None:
    """Two runs over identical cells return identical frozen results."""
    runner = SweepRunner(party=PARTY, workers=2)
    cells = _cells()

    first = runner.run(cells)
    second = runner.run(cells)

    assert first == second


def test_parallel_equivalence_workers_1_vs_4() -> None:
    """workers=1 and workers=4 produce identical results (thread-safe registry)."""
    cells = _cells()

    serial = SweepRunner(party=PARTY, workers=1).run(cells)
    parallel = SweepRunner(party=PARTY, workers=4).run(cells)

    assert serial == parallel


def test_cost_bound_raises_on_budget_exceeded() -> None:
    """Cells summing past max_total_runs raise a ValueError naming runs/budget."""
    runner = SweepRunner(party=PARTY, workers=2)
    cells = (
        SweepCell(
            scenario="a", strategy="s", spec=VARIANT_A, policy=None, runs=3, seed=1
        ),
        SweepCell(
            scenario="b", strategy="s", spec=VARIANT_B, policy=None, runs=3, seed=1
        ),
    )

    with pytest.raises(ValueError, match="runs") as excinfo:
        runner.run(cells, max_total_runs=5)

    assert "budget" in str(excinfo.value)


def test_validation_workers_party_runs() -> None:
    """workers=0, an empty party, and a runs=0 cell all raise ValueError."""
    with pytest.raises(ValueError, match="workers"):
        SweepRunner(party=PARTY, workers=0)
    with pytest.raises(ValueError, match="party"):
        SweepRunner(party="", workers=2)

    runner = SweepRunner(party=PARTY, workers=2)
    bad = SweepCell(
        scenario="a", strategy="s", spec=VARIANT_A, policy=None, runs=0, seed=1
    )
    with pytest.raises(ValueError, match="runs"):
        runner.run((bad,))


def test_policy_none_matches_direct_run_encounter() -> None:
    """A None-policy cell equals a direct run_encounter on the same scenario."""
    runner = SweepRunner(party=PARTY, workers=2)
    cell = SweepCell(
        scenario="variant_a",
        strategy="default",
        spec=VARIANT_A,
        policy=None,
        runs=RUNS,
        seed=SEED,
    )

    (result,) = runner.run((cell,))
    report, outcomes = run_encounter(VARIANT_A, RUNS, SEED, collect_runs=True)

    assert result.report == report
    assert result.danger == danger_metrics(outcomes, PARTY)


def test_format_sweep_columns_and_win_pct() -> None:
    """The renderer shows header tokens, one row per result, and the party win%."""
    runner = SweepRunner(party=PARTY, workers=2)
    results = runner.run_grid(SCENARIOS, STRATEGIES, runs=RUNS, seed=SEED)

    text = format_sweep(results, PARTY)
    for token in ("Scenario", "Win%", "P(TPK)"):
        assert token in text

    lines = text.splitlines()
    assert len(lines) == 2 + len(results)

    first = results[0]
    expected_win = f"{first.report.wins.get(PARTY, 0) / first.report.runs * 100:.1f}"
    assert lines[2].split()[3] == expected_win