# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-agent-checkpoint-reject-a-replay-that-reproduces-byte-identical-state.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-agent-checkpoint-reject-a-replay-that-reproduces-byte-identical-state.warrant.json
# Markdown SHA-256:   7b93b39f15efe18be29ebab3b8f14634b83f097d8caafa9a054b00e495788557
# Code AST SHA-256:   442e5e068a190c86774fef4e439c66fd384d98400ab23564d366290647832446
# Verified Classes:   ReplayDivergenceError, Observation, Checkpoint
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field
from typing import List, Tuple


class ReplayDivergenceError(RuntimeError):
    """Raised when a checkpoint replay fails observational equality
    even though post-state equality holds (or cannot be proven)."""
    pass


@dataclass(frozen=True)
class Observation:
    step: int
    source: str
    payload_digest: str
    pinned_at: float | None  # None => re-fetched, provenance unprovable


@dataclass
class Checkpoint:
    step: int
    state_digest: str
    observations: List[Observation] = field(default_factory=list)


def _digest_sequence(obs: List[Observation]) -> str:
    h = hashlib.sha256()
    for o in sorted(obs, key=lambda x: (x.step, x.source)):
        h.update(f"{o.step}|{o.source}|{o.payload_digest}|{o.pinned_at}".encode())
    return h.hexdigest()


def validate_replay(recorded: Checkpoint, replayed: Checkpoint) -> None:
    # Condition 1: state equality.
    if recorded.state_digest != replayed.state_digest:
        raise ReplayDivergenceError(
            f"state divergence at step {replayed.step}: "
            f"{recorded.state_digest[:8]} != {replayed.state_digest[:8]}"
        )

    # Condition 3: every nondeterministic read must be pinned.
    for o in replayed.observations:
        if o.pinned_at is None:
            raise ReplayDivergenceError(
                f"unpinned observation at step {o.step} source={o.source}; "
                "observational equivalence cannot be proven"
            )

    # Condition 2: observational equality. State equality must NOT override this.
    if _digest_sequence(recorded.observations) != _digest_sequence(replayed.observations):
        raise ReplayDivergenceError(
            f"observational divergence at step {replayed.step}: "
            "state matched but observation sequence differs; "
            "recovery requires operator sign-off"
        )
    # Accepted: resume.