From 6c6de16e60a137d6edab8f23574820e8cc5a31b1 Mon Sep 17 00:00:00 2001 From: ctan Date: Mon, 21 Sep 2026 15:08:34 +0200 Subject: [PATCH] feat(phase2): replay log and replay policy (P0.5) Add replay.py: a ReplayLog records every (observation, action) pair a policy produced, keyed by monotonic call_seq, plus the EncounterSpec, seed and round_cap, and serializes the whole battle to JSONL. replay() turns a log back into an observation-only LLMPolicy that returns each recorded action, raising ReplayMismatchError on a differing observation and ReplayExhaustedError past the last call. Observations are matched by their canonical JSON (observation_key), not raw dataclass equality, because JSON collapses tuples and lists and the snapshot's inner dict[str, object] holds both. Action and the spec are dumped through Pydantic TypeAdapters (mode=json) so tuple fields survive as arrays and are coerced back on load. Move TrickCondition/ReadyTrigger out of the TYPE_CHECKING block in combat.py: TypeAdapter(Action) must resolve them at runtime to serialize the dataclass, so the Literal aliases now live at module level. --- src/pf1e_simulator/combat.py | 18 +-- src/pf1e_simulator/replay.py | 250 +++++++++++++++++++++++++++++++++++ tests/test_phase2_replay.py | 234 ++++++++++++++++++++++++++++++++ 3 files changed, 494 insertions(+), 8 deletions(-) create mode 100644 src/pf1e_simulator/replay.py create mode 100644 tests/test_phase2_replay.py diff --git a/src/pf1e_simulator/combat.py b/src/pf1e_simulator/combat.py index c226a15..000df26 100644 --- a/src/pf1e_simulator/combat.py +++ b/src/pf1e_simulator/combat.py @@ -178,19 +178,21 @@ if TYPE_CHECKING: from pf1e_simulator.rng import Rng from pf1e_simulator.spells import SpellSpec - TrickCondition = Literal[ - "dazzled", "blinded", "deafened", "entangled", "shaken", "sickened" - ] - - ReadyTrigger = Literal[ - "enemy_enters_reach", "enemy_casts", "enemy_attacks", "enemy_hits_ally" - ] - from pf1e_simulator.los import ( cover_bonus_cells, has_line_of_effect_cells, ) +# Runtime aliases (not only annotations): Pydantic type adapters over ``Action`` +# must resolve them, so they cannot live inside the TYPE_CHECKING block. +TrickCondition = Literal[ + "dazzled", "blinded", "deafened", "entangled", "shaken", "sickened" +] + +ReadyTrigger = Literal[ + "enemy_enters_reach", "enemy_casts", "enemy_attacks", "enemy_hits_ally" +] + _SQUARE_FT = 5 # Phase 0 maps use 5-ft squares _NATURAL_ONE = 1 # PF1e: natural 1 always misses _NATURAL_TWENTY = 20 # PF1e: natural 20 always hits and threatens diff --git a/src/pf1e_simulator/replay.py b/src/pf1e_simulator/replay.py new file mode 100644 index 0000000..760bca7 --- /dev/null +++ b/src/pf1e_simulator/replay.py @@ -0,0 +1,250 @@ +"""Replay log for observation-only policies (docs/phase-2 P0.5). + +A ``ReplayLog`` records every ``(observation, action)`` pair an +observation-only policy produced during one battle, keyed by a monotonic +``call_seq`` assigned in execution order. It also carries the ``EncounterSpec`` +plus the battle ``seed`` and ``round_cap`` so a recorded battle can be re-run +standalone. ``replay(log)`` turns a log back into an ``LLMPolicy`` that returns +the recorded action for each observation, raising a typed ``ReplayError`` (a +``ValueError``) on an out-of-sequence observation or once the log is exhausted. + +Why ``call_seq`` and not the observation as the key: the same observation can +legitimately recur (a stalled battle), and delayed turns / summons can revisit +an earlier position — the sequence position is the only unambiguous key. + +The JSONL serialization is self-contained (a header line with the spec, seed +and round cap, then one record per line) so ``ReplayLog.from_jsonl(text)`` +round-trips without any out-of-band state. ``Observation`` uses the engine's +canonical ``to_dict()`` (``schema_version`` included); ``Action`` and the spec +are dumped through Pydantic type adapters so their tuple fields survive as +JSON arrays and are coerced back to tuples on load. Observations are matched +during playback by :func:`observation_key` (their canonical JSON), not raw +dataclass equality: JSON collapses tuples and lists, and the snapshot's inner +``dict[str, object]`` holds both. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import TYPE_CHECKING, Self, TypeGuard + +from pydantic import TypeAdapter + +from pf1e_simulator.combat import Action, Observation +from pf1e_simulator.map import MapSpec +from pf1e_simulator.models import Combatant +from pf1e_simulator.runner import EncounterSpec + +if TYPE_CHECKING: + from pf1e_simulator.combat import LLMPolicy + +_HEADER_KIND = "header" +_RECORD_KIND = "record" + +# ``EncounterSpec`` is a stdlib dataclass whose annotations reference +# ``MapSpec``/``Combatant``; ``runner`` imports those only under TYPE_CHECKING, +# so the adapter's default namespace cannot resolve them. Rebuild it with the +# two runtime classes in scope. +_SPEC_ADAPTER = TypeAdapter(EncounterSpec) +_SPEC_ADAPTER.rebuild(_types_namespace={"MapSpec": MapSpec, "Combatant": Combatant}) +_OBSERVATION_ADAPTER = TypeAdapter(Observation) +_ACTION_ADAPTER = TypeAdapter(Action) + + +class ReplayError(ValueError): + """Base error for replay-log construction and playback.""" + + +class ReplayMismatchError(ReplayError): + """A replay policy received an observation that differs from the recorded one.""" + + +class ReplayExhaustedError(ReplayError): + """A replay policy was called past the last recorded ``call_seq``.""" + + +@dataclass(frozen=True) +class ReplayRecord: + """One recorded policy call: the snapshot shown and the action chosen. + + ``call_seq`` is the position of this call in the execution order of the + battle (0-based, contiguous), assigned by the recorder. + """ + + call_seq: int + observation: Observation + action: Action + + +class ReplayLog: + """A recorded battle: spec + seed + round cap + ordered policy calls.""" + + def __init__(self, *, spec: EncounterSpec, seed: int, round_cap: int) -> None: + self.spec = spec + self.seed = seed + self.round_cap = round_cap + self._records: list[ReplayRecord] = [] + + def add(self, record: ReplayRecord) -> None: + """Append one call; require a strictly increasing ``call_seq``. + + The log is a sequence: a duplicate or out-of-order ``call_seq`` would + make the recorded execution order ambiguous, so both are rejected. + """ + if self._records: + last = self._records[-1].call_seq + if record.call_seq == last: + msg = f"duplicate call_seq {record.call_seq}" + raise ReplayError(msg) + if record.call_seq < last: + msg = f"call_seq {record.call_seq} is not after {last}" + raise ReplayError(msg) + self._records.append(record) + + @property + def records(self) -> tuple[ReplayRecord, ...]: + """The recorded calls ordered by ``call_seq`` (replay order).""" + return tuple(sorted(self._records, key=lambda record: record.call_seq)) + + def __len__(self) -> int: + return len(self._records) + + def to_jsonl(self) -> str: + """Serialize to newline-delimited JSON (header line, then one per call).""" + lines = [self._header_json()] + lines.extend(self._record_json(record) for record in self.records) + return "".join(line + "\n" for line in lines) + + @classmethod + def from_jsonl(cls, text: str) -> Self: + """Rebuild a log from ``to_jsonl`` output; raise ``ReplayError`` if invalid.""" + lines = [line for line in text.splitlines() if line.strip()] + if not lines: + msg = "empty replay log" + raise ReplayError(msg) + header = _as_object(_loads(lines[0]), "header") + if _as_str(header.get("kind"), "kind") != _HEADER_KIND: + msg = "first line is not a replay header" + raise ReplayError(msg) + log = cls( + spec=_SPEC_ADAPTER.validate_python(header["spec"]), + seed=_as_int(header.get("seed"), "seed"), + round_cap=_as_int(header.get("round_cap"), "round_cap"), + ) + for line in lines[1:]: + data = _as_object(_loads(line), "record") + if _as_str(data.get("kind"), "kind") != _RECORD_KIND: + msg = "record line is not a replay record" + raise ReplayError(msg) + log.add( + ReplayRecord( + call_seq=_as_int(data.get("call_seq"), "call_seq"), + observation=_OBSERVATION_ADAPTER.validate_python( + data["observation"] + ), + action=_ACTION_ADAPTER.validate_python(data["action"]), + ) + ) + return log + + def _header_json(self) -> str: + payload: dict[str, object] = { + "kind": _HEADER_KIND, + "seed": self.seed, + "round_cap": self.round_cap, + "spec": _SPEC_ADAPTER.dump_python(self.spec, mode="json"), + } + return json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ) + + def _record_json(self, record: ReplayRecord) -> str: + payload: dict[str, object] = { + "kind": _RECORD_KIND, + "call_seq": record.call_seq, + "observation": record.observation.to_dict(), + "action": _ACTION_ADAPTER.dump_python( + record.action, mode="json", exclude_defaults=True + ), + } + return json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ) + + +def observation_key(observation: Observation) -> str: + """Return the canonical identity of an observation for replay matching. + + ``Observation.to_dict()`` is already JSON-shaped, so dumping it with sorted + keys yields a value that is insensitive to dict insertion order and treats + tuples and lists alike (JSON has no tuple type). + """ + return json.dumps( + observation.to_dict(), sort_keys=True, separators=(",", ":"), ensure_ascii=True + ) + + +def replay(log: ReplayLog) -> LLMPolicy: + """Build an observation-only policy that replays ``log`` in ``call_seq`` order. + + The returned policy walks the log once, comparing the observation it is + handed to the recorded one before returning that call's action (wrapped in + a one-action tuple). An out-of-sequence observation raises + ``ReplayMismatchError``; a call past the end raises ``ReplayExhaustedError``. + """ + ordered = log.records + keys = [observation_key(record.observation) for record in ordered] + cursor = 0 + + def policy(observation: Observation) -> tuple[Action, ...]: + nonlocal cursor + if cursor >= len(ordered): + msg = f"replay log exhausted after {len(ordered)} calls" + raise ReplayExhaustedError(msg) + record = ordered[cursor] + if keys[cursor] != observation_key(observation): + msg = ( + f"replay mismatch at call_seq {record.call_seq}: " + "the observation differs from the recorded one" + ) + raise ReplayMismatchError(msg) + cursor += 1 + return (record.action,) + + return policy + + +def _loads(text: str) -> object: + """Parse one JSONL line, mapping a JSON error to a ``ReplayError``.""" + try: + return json.loads(text) + except json.JSONDecodeError as exc: + msg = f"invalid JSON in replay log: {exc}" + raise ReplayError(msg) from exc + + +def _as_object(value: object, what: str) -> dict[str, object]: + if _is_str_object_dict(value): + return value + msg = f"expected a JSON object for {what}" + raise ReplayError(msg) + + +def _as_str(value: object, what: str) -> str: + if isinstance(value, str): + return value + msg = f"expected a string for {what}" + raise ReplayError(msg) + + +def _as_int(value: object, what: str) -> int: + # bool is an int subclass but never a valid call_seq/seed/round_cap. + if isinstance(value, int) and not isinstance(value, bool): + return value + msg = f"expected an integer for {what}" + raise ReplayError(msg) + + +def _is_str_object_dict(value: object) -> TypeGuard[dict[str, object]]: + return isinstance(value, dict) diff --git a/tests/test_phase2_replay.py b/tests/test_phase2_replay.py new file mode 100644 index 0000000..c915a7c --- /dev/null +++ b/tests/test_phase2_replay.py @@ -0,0 +1,234 @@ +"""Phase-2 replay log (docs/phase-2-llm-architecture.md P0.5). + +``replay.py`` records every ``(observation, action)`` pair a policy produced, +keyed by the execution-order ``call_seq``, and serializes the whole battle +(spec + seed + round cap + calls) to JSONL. ``replay()`` turns the log back +into an observation-only policy returning the recorded actions. + +These tests pin the contract: JSONL round-trips to canonically-equal records +(``observation_key`` ignores JSON's tuple/list collapse), a replayed battle is +bit-identical to the recorded one (including from a serialized log), a +duplicate ``call_seq`` is rejected, and playback fails loudly on a mismatched +observation and once the log is exhausted. +""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path +from typing import TYPE_CHECKING + +import pytest + +from pf1e_simulator.candidates import candidates +from pf1e_simulator.loaders import load_combatant +from pf1e_simulator.map import load_map +from pf1e_simulator.replay import ( + ReplayError, + ReplayExhaustedError, + ReplayLog, + ReplayMismatchError, + ReplayRecord, + observation_key, + replay, +) +from pf1e_simulator.runner import EncounterSpec, Side, run_encounter +from pf1e_simulator.spells import SPELLS_DIR, load_spell_registry + +if TYPE_CHECKING: + from pf1e_simulator.combat import Action, CombatantState, Observation + from pf1e_simulator.models import Combatant + from pf1e_simulator.runner import EncounterReport + +ROOT = Path(__file__).resolve().parents[1] +MAP = ROOT / "data" / "maps" / "sample_arena.yaml" +SEED = 7 + + +def _c(path: str) -> Combatant: + return load_combatant(ROOT / "data" / path) + + +def _duel_spec() -> EncounterSpec: + """A fast 1v1 (goblin archer vs orc) on the sample arena.""" + return EncounterSpec( + map=load_map(MAP), + round_cap=20, + sides=( + Side(name="players", combatants=(_c("monsters/goblin.json"),), zone="A"), + Side(name="monsters", combatants=(_c("monsters/orc.json"),), zone="B"), + ), + ) + + +class _Recorder: + """Legacy two-argument policy: log every call, then pick ``candidates[0]``.""" + + def __init__(self, spec: EncounterSpec, seed: int, round_cap: int) -> None: + self.log = ReplayLog(spec=spec, seed=seed, round_cap=round_cap) + + def policy( + self, observation: Observation, _state: CombatantState + ) -> tuple[Action, ...]: + action = candidates(observation)[0] + self.log.add( + ReplayRecord( + call_seq=len(self.log), + observation=observation, + action=action, + ) + ) + return (action,) + + +def _record(spec: EncounterSpec, *, seed: int = SEED) -> tuple[_Recorder, EncounterReport]: + """Record one battle and return the recorder plus its captured report.""" + registry = load_spell_registry(SPELLS_DIR) + recorder = _Recorder(spec, seed=seed, round_cap=spec.round_cap) + report = run_encounter( + spec, + runs=1, + seed=seed, + llm_policy=recorder.policy, + spell_registry=registry, + capture_first_transcript=True, + ) + return recorder, report + + +def test_jsonl_round_trip_preserves_records() -> None: + """A log serializes to JSONL and loads back to structurally-equal records.""" + spec = _duel_spec() + recorder, _report = _record(spec) + log = recorder.log + + assert len(log) > 0 + assert [record.call_seq for record in log.records] == list(range(len(log))) + + text = log.to_jsonl() + restored = ReplayLog.from_jsonl(text) + + assert len(restored) == len(log) + assert [record.call_seq for record in restored.records] == [ + record.call_seq for record in log.records + ] + assert [ + observation_key(record.observation) for record in restored.records + ] == [observation_key(record.observation) for record in log.records] + assert [record.action for record in restored.records] == [ + record.action for record in log.records + ] + assert restored.spec == log.spec + assert restored.seed == log.seed + assert restored.round_cap == log.round_cap + assert restored.to_jsonl() == text + + +def test_replay_reproduces_the_recorded_battle() -> None: + """Replaying the in-memory log matches the recorded battle exactly.""" + spec = _duel_spec() + registry = load_spell_registry(SPELLS_DIR) + recorder, recorded = _record(spec) + + replayed = run_encounter( + spec, + runs=1, + seed=SEED, + llm_policy=replay(recorder.log), + spell_registry=registry, + capture_first_transcript=True, + ) + + assert replayed.wins == recorded.wins + assert replayed.draws == recorded.draws + assert replayed.avg_rounds == recorded.avg_rounds + assert replayed.transcript == recorded.transcript + + +def test_replay_from_serialized_log_reproduces_the_battle() -> None: + """A log thawed from JSONL replays the battle bit-for-bit.""" + spec = _duel_spec() + registry = load_spell_registry(SPELLS_DIR) + recorder, recorded = _record(spec) + + restored = ReplayLog.from_jsonl(recorder.log.to_jsonl()) + replayed = run_encounter( + spec, + runs=1, + seed=SEED, + llm_policy=replay(restored), + spell_registry=registry, + capture_first_transcript=True, + ) + + assert replayed.wins == recorded.wins + assert replayed.draws == recorded.draws + assert replayed.avg_rounds == recorded.avg_rounds + assert replayed.transcript == recorded.transcript + + +def test_replay_rejects_a_mismatched_observation() -> None: + """An observation that differs from the recorded one raises a mismatch error.""" + spec = _duel_spec() + recorder, _report = _record(spec) + log = recorder.log + policy = replay(log) + first = log.records[0].observation + tampered = replace(first, round=first.round + 1) + + with pytest.raises(ReplayMismatchError): + policy(tampered) + + +def test_replay_returns_each_recorded_action_then_exhausts() -> None: + """Playback walks every record in order, then raises once past the end.""" + spec = _duel_spec() + recorder, _report = _record(spec) + log = recorder.log + policy = replay(log) + + for record in log.records: + assert policy(record.observation) == (record.action,) + + with pytest.raises(ReplayExhaustedError): + policy(log.records[-1].observation) + + +def test_replay_log_rejects_a_duplicate_call_seq() -> None: + """Adding two records with the same ``call_seq`` is an ambiguity error.""" + spec = _duel_spec() + recorder, _report = _record(spec) + record = recorder.log.records[-1] + + with pytest.raises(ReplayError, match="duplicate call_seq"): + recorder.log.add( + ReplayRecord( + call_seq=record.call_seq, + observation=record.observation, + action=record.action, + ) + ) + + +def test_replay_log_rejects_an_out_of_order_call_seq() -> None: + """A ``call_seq`` below the last one is an ambiguity error (not a duplicate).""" + spec = _duel_spec() + recorder, _report = _record(spec) + record = recorder.log.records[0] + + with pytest.raises(ReplayError, match="is not after"): + recorder.log.add( + ReplayRecord( + call_seq=record.call_seq, + observation=record.observation, + action=record.action, + ) + ) + + +def test_from_jsonl_rejects_garbage() -> None: + """An empty or unparseable payload raises ``ReplayError``, not a raw error.""" + with pytest.raises(ReplayError): + ReplayLog.from_jsonl("") + with pytest.raises(ReplayError): + ReplayLog.from_jsonl("{not json}")