PIG-Demo/envs/verify_fixtures.py

#!/usr/bin/env python3
"""Replay every committed fixture and reproduce its recorded rewards.

A fixture is a claim: "this model, on this seed, scored this." The claim is
only worth anything if it can be re-derived from the moves it records. This
does that in Python; `verify.ts` does the same thing in the browser, live, in
front of the visitor.

Exits non-zero if any fixture's rewards cannot be reproduced from its own turns.
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent / "wordle_five"))
sys.path.insert(0, str(Path(__file__).parent / "alert_triage"))

from alert_triage import taskset as AT  # noqa: E402
from alert_triage.generator import is_held_out  # noqa: E402
from wordle_five import solver as S  # noqa: E402
from wordle_five.engine import Game  # noqa: E402
from wordle_five.protocol import parse_guess  # noqa: E402
from wordle_five.reward import Episode, score  # noqa: E402

TRACES = Path(__file__).parent.parent / "public" / "traces"
TOLERANCE = 1e-9


def replay_wordle(fixture: dict) -> dict[str, float]:
    game = Game(seed=fixture["seed"])
    if game.answer != fixture["answer"]:
        raise AssertionError(
            f"seed {fixture['seed']} gives {game.answer!r}, fixture claims {fixture['answer']!r}"
        )
    for turn in fixture["turns"]:
        guess = parse_guess(turn["reply"])
        if guess is None:
            game.rejected += 1
            continue
        game.play(guess)

    episode = Episode(
        answer=game.answer,
        guesses=[g for g, _ in game.history],
        patterns=[p for _, p in game.history],
        rejected=game.rejected,
        reference_depth=S.reference_depth(game.answer),
    )
    return score(episode)


def replay_alert_triage(fixture: dict) -> dict[str, float | None]:
    """The world comes from the seed, the score from the reply strings. Nothing
    else in the fixture is trusted — the same rule the browser's verify uses."""
    seed = fixture["seed"]
    if is_held_out(seed):
        raise AssertionError(f"seed {seed} is in the held-out bucket and must never be captured")
    played = AT.replay(seed, [turn["reply"] for turn in fixture["turns"]])
    if played["outcome"] != fixture["outcome"]:
        raise AssertionError(f"replay gives outcome {played['outcome']!r}, fixture claims {fixture['outcome']!r}")
    return played["rewards"]


REPLAY = {"wordle": replay_wordle, "alert-triage": replay_alert_triage}


def _delta(recomputed, recorded) -> float:
    """None is "not scored". It matches only None; against a number it is a mismatch, not a zero."""
    if recomputed is None or recorded is None:
        return 0.0 if recomputed is recorded else float("inf")
    return abs(recomputed - recorded)


def main() -> int:
    fixtures = [
        f for demo in sorted(p for p in TRACES.iterdir() if p.is_dir())
        for f in sorted(demo.glob("*.json"))
    ]
    if not fixtures:
        print("no fixtures found — nothing to verify")
        return 0

    failures = 0
    for path in fixtures:
        demo = path.parent.name
        if demo not in REPLAY:
            failures += 1
            print(f"{path.parent.name}/{path.name:<20}  NO REPLAYER for demo '{demo}'")
            continue
        data = json.loads(path.read_text())
        try:
            recomputed = REPLAY[demo](data)
        except AssertionError as exc:
            failures += 1
            print(f"{path.parent.name}/{path.name:<20}  MISMATCH  {exc}")
            continue
        recorded = data["rewards"]
        deltas = {k: _delta(recomputed[k], recorded[k]) for k in recorded}
        worst = max(deltas.values())
        status = "ok" if worst <= TOLERANCE else "MISMATCH"
        if worst > TOLERANCE:
            failures += 1
            culprit = max(deltas, key=deltas.get)
            print(f"{path.parent.name}/{path.name:<20}  {status}  worst delta {worst:.3g} on '{culprit}'")
        else:
            print(f"{path.parent.name}/{path.name:<20}  {status}  delta {worst:.1e}")

    print(f"\n{len(fixtures)} fixtures, {failures} mismatched")
    return 1 if failures else 0


if __name__ == "__main__":
    raise SystemExit(main())