# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-memory-poisoning-in-read-write-forget-persistent-agent-stores.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-memory-poisoning-in-read-write-forget-persistent-agent-stores.warrant.json
# Markdown SHA-256:   144885d962f44fcdea2fc92c4dbed9f22b24b9ddd8435c50ea9691d49250fd31
# Code AST SHA-256:   9aaae61906c02237dc92ff2f85180854e3b3c467eec762712c6cb6beeadda0c2
# Verified Classes:   MemoryDivergenceError, LedgerEntry, DivergenceDetector
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class MemoryDivergenceError(RuntimeError):
    """Raised when store state diverges from the instance's own write ledger."""


@dataclass(frozen=True)
class LedgerEntry:
    version: int
    content_hash: str
    in_flight: bool = False


def _hash(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


class DivergenceDetector:
    def __init__(self, holds_exclusive_lease: bool):
        self._ledger: dict[str, LedgerEntry] = {}
        self._lease = holds_exclusive_lease

    def record_write(self, key: str, version: int, value: bytes) -> None:
        # Written before the store acknowledges the write to callers.
        self._ledger[key] = LedgerEntry(version, _hash(value), in_flight=True)

    def ack_write(self, key: str, version: int, value: bytes) -> None:
        self._ledger[key] = LedgerEntry(version, _hash(value), in_flight=False)

    def check_read(self, key: str, version: int, value: bytes) -> None:
        observed = _hash(value)
        entry = self._ledger.get(key)
        if entry is None:
            return  # never written by us; not our boundary
        if entry.in_flight:
            return  # unacknowledged write window; defer
        if observed == entry.content_hash:
            if version != entry.version:
                # benign rewrite of identical content
                self._ledger[key] = LedgerEntry(version, observed, in_flight=False)
            return
        if version == entry.version and observed != entry.content_hash:
            if self._lease:
                raise MemoryDivergenceError(
                    f"content changed at fixed version {version} for key {key!r}; "
                    f"expected {entry.content_hash[:12]}, got {observed[:12]}"
                )
            return  # indeterminate: no exclusive lease
        if version > entry.version:
            raise MemoryDivergenceError(
                f"version advanced {entry.version}->{version} with foreign content "
                f"for key {key!r}"
            )

    def verify_forget(self, key: str, still_present: bool) -> None:
        if still_present:
            raise MemoryDivergenceError(
                f"forget of key {key!r} reported success but key still readable; "
                f"deletion failed under replication lag"
            )