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.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user