# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/pending-what-compare-and-swap-and-version-preconditions-must-an-instance-guard-read-modify-write-cycles-to-prevent-lost-updates.md
# Remnant Warrant:    https://corpusfetch.com/c/pending-what-compare-and-swap-and-version-preconditions-must-an-instance-guard-read-modify-write-cycles-to-prevent-lost-updates.warrant.json
# Markdown SHA-256:   ac18555a3c137d179b47367dc90e7a846cbf6a919707188ab7e2e611fa69f891
# Code AST SHA-256:   4a9e6ca721624b264cfbc29da1dce2cc33dc01640ec3ee1adff582580180772f
# Verified Classes:   OptimisticConcurrencyConflictError, VersionedRecord, CompareAndSwapVersionGate
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Any


class OptimisticConcurrencyConflictError(RuntimeError):
    """Raised when a Compare-And-Swap (CAS) version precondition fails during a write commit."""
    pass


@dataclass(frozen=True)
class VersionedRecord:
    record_id: str
    version: int
    payload: dict[str, Any]


class CompareAndSwapVersionGate:
    """
    Prevents lost updates across concurrent read-modify-write (RMW) cycles:
    1. Requires every mutation to supply the exact `expected_version` observed during read.
    2. Rejects blind overwrites when the in-store version has advanced.
    3. Atomically increments the monotonic version counter on successful commit.
    """
    def __init__(self) -> None:
        self._store: dict[str, VersionedRecord] = {}

    def create_initial(self, record_id: str, payload: dict[str, Any]) -> VersionedRecord:
        if record_id in self._store:
            raise OptimisticConcurrencyConflictError(f"Record {record_id!r} already exists.")
        rec = VersionedRecord(record_id=record_id, version=1, payload=dict(payload))
        self._store[record_id] = rec
        return rec

    def read_snapshot(self, record_id: str) -> VersionedRecord:
        if record_id not in self._store:
            raise KeyError(f"Record {record_id!r} not found.")
        return self._store[record_id]

    def compare_and_swap(
        self,
        record_id: str,
        expected_version: int,
        new_payload: dict[str, Any],
    ) -> VersionedRecord:
        current = self._store.get(record_id)
        if current is None:
            raise KeyError(f"Record {record_id!r} does not exist.")
        if current.version != expected_version:
            raise OptimisticConcurrencyConflictError(
                f"CAS precondition failed for {record_id!r}: expected v{expected_version}, found v{current.version}."
            )
        updated = VersionedRecord(
            record_id=record_id,
            version=current.version + 1,
            payload=dict(new_payload),
        )
        self._store[record_id] = updated
        return updated