# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-emit-a-memory-mutation-audit-event-before-the-mutation-becomes-observable.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-emit-a-memory-mutation-audit-event-before-the-mutation-becomes-observable.warrant.json
# Markdown SHA-256:   b1138cca9b10d6af8d3022974795ea0087efadb22b6e4edaed615d6b6b4b8879
# Code AST SHA-256:   bcd50976d43a5e70b3fb4ab5b148453ad9f8fba1970fc7225c2f42427d7eb726
# Verified Classes:   AuditOrderingViolation, AuditSinkUnavailable, AuditRecord, WriteAheadMemory
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


class AuditOrderingViolation(RuntimeError):
    """Raised when a mutated value becomes readable before its audit event is durable."""
    pass


class AuditSinkUnavailable(RuntimeError):
    """Raised when the durable audit log cannot accept a record."""
    pass


@dataclass
class AuditRecord:
    seq: int
    key: str
    old_value: Any
    new_value: Any
    ts: float


@dataclass
class WriteAheadMemory:
    """Memory store enforcing record-before-visible ordering."""
    durable_audit_log: List[AuditRecord] = field(default_factory=list)
    _visible: Dict[str, Any] = field(default_factory=dict)
    _pending: Dict[str, Any] = field(default_factory=dict)
    _seq: int = 0
    _audit_accepting: bool = True

    def _append_durable(self, key: str, old: Any, new: Any) -> AuditRecord:
        if not self._audit_accepting:
            raise AuditSinkUnavailable(
                f"audit sink rejected record for key={key!r}; mutation must not become visible"
            )
        self._seq += 1
        rec = AuditRecord(self._seq, key, old, new, time.time())
        # Simulated synchronous durability: log append completes before return.
        self.durable_audit_log.append(rec)
        return rec

    def write(self, key: str, value: Any) -> None:
        old = self._visible.get(key)
        # 1) Durable record first.
        self._append_durable(key, old, value)
        # 2) Only now publish to the readable store.
        self._visible[key] = value

    def read(self, key: str) -> Any:
        if key in self._pending:
            raise AuditOrderingViolation(
                f"read of key={key!r} observed a value with no durable audit record"
            )
        return self._visible.get(key)

    def replay(self) -> Dict[str, Any]:
        """Reconstruct state from the audit log ALONE. Never reads _visible."""
        state: Dict[str, Any] = {}
        for rec in self.durable_audit_log:
            state[rec.key] = rec.new_value
        # Divergence check: any readable value lacking a producing record is a violation.
        for k, v in self._visible.items():
            if k not in state:
                raise AuditOrderingViolation(
                    f"replay divergence: key={k!r} readable but has no audit event"
                )
        return state


def safe_write(store: WriteAheadMemory, key: str, value: Any) -> None:
    try:
        store.write(key, value)
    except AuditSinkUnavailable as exc:
        # Fail closed: do not publish; surface as ordering violation to caller.
        raise AuditOrderingViolation(
            f"refusing to mutate {key!r}: audit durability not established ({exc})"
        ) from exc