# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/along-what-monotonicity-and-quorum-invariants-must-an-instance-reconcile-divergent-state-replicas-across-asynchronous-tool-caches.md
# Remnant Warrant:    https://corpusfetch.com/c/along-what-monotonicity-and-quorum-invariants-must-an-instance-reconcile-divergent-state-replicas-across-asynchronous-tool-caches.warrant.json
# Markdown SHA-256:   d0cb87473e128625a7768225b3ca188e85d8cd3f4c85de54c6282719664c5f19
# Code AST SHA-256:   7f61760aa08e0493a3a31d75155b2276f5553f264602c53950acc388fd79aa04
# Verified Classes:   QuorumOrMonotonicityError, ReplicaStateSample, ReconciledQuorumState, MonotonicQuorumReconciler
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Any


class QuorumOrMonotonicityError(RuntimeError):
    '''Raised when replica responses fail quorum intersection, violate monotonic reads, or exhibit split-brain.'''
    pass


@dataclass(frozen=True)
class ReplicaStateSample:
    replica_id: str
    entity_id: str
    epoch: int
    version: int
    payload_sha256: str
    payload: dict[str, Any]


@dataclass(frozen=True)
class ReconciledQuorumState:
    entity_id: str
    winning_epoch: int
    winning_version: int
    winning_sha256: str
    authoritative_payload: dict[str, Any]
    stale_replica_ids_for_repair: tuple[str, ...]


class MonotonicQuorumReconciler:
    '''
    Reconciles divergent state replicas across asynchronous tool caches:
    1. Enforces Strict Quorum Intersection (`len(samples) + write_quorum > total_replicas`).
    2. Detects split-brain corruption when identical `(epoch, version)` coordinates carry different SHA-256 digests.
    3. Enforces session Monotonic Reads against `_session_watermarks`.
    4. Identifies lagging replicas requiring read-repair or cache invalidation.
    '''

    def __init__(self) -> None:
        self._session_watermarks: dict[str, tuple[int, int]] = {}

    def reconcile(
        self,
        *,
        entity_id: str,
        total_replicas: int,
        write_quorum: int,
        samples: list[ReplicaStateSample],
    ) -> ReconciledQuorumState:
        if total_replicas <= 0 or write_quorum <= 0 or write_quorum > total_replicas:
            raise ValueError("Invalid total_replicas or write_quorum configuration.")

        # Gate 1: Strict Quorum Intersection (R + W > N)
        unique_replicas = {s.replica_id: s for s in samples if s.entity_id == entity_id}
        r_count = len(unique_replicas)
        if r_count + write_quorum <= total_replicas:
            raise QuorumOrMonotonicityError(
                f"Quorum Intersection Failed for '{entity_id}': R={r_count} + W={write_quorum} <= N={total_replicas}."
            )

        # Gate 2: Split-brain detection at identical (epoch, version)
        coord_hashes: dict[tuple[int, int], str] = {}
        for sample in unique_replicas.values():
            coord = (sample.epoch, sample.version)
            if coord in coord_hashes and coord_hashes[coord] != sample.payload_sha256:
                raise QuorumOrMonotonicityError(
                    f"Split-Brain Detected on '{entity_id}' at (epoch={sample.epoch}, version={sample.version}): "
                    f"digest '{coord_hashes[coord]}' != '{sample.payload_sha256}'."
                )
            coord_hashes[coord] = sample.payload_sha256

        # Gate 3: Select maximum (epoch, version) and enforce Session Monotonic Reads
        winner = max(unique_replicas.values(), key=lambda s: (s.epoch, s.version))
        win_coord = (winner.epoch, winner.version)
        last_seen = self._session_watermarks.get(entity_id, (-1, -1))

        if win_coord < last_seen:
            raise QuorumOrMonotonicityError(
                f"Monotonic Read Violation on '{entity_id}': quorum max version {win_coord} "
                f"regressed behind session high-water mark {last_seen}."
            )

        # Gate 4: Advance watermark and list lagging replicas for read-repair
        self._session_watermarks[entity_id] = win_coord
        stale_ids = tuple(
            sorted(s.replica_id for s in unique_replicas.values() if (s.epoch, s.version) < win_coord)
        )

        return ReconciledQuorumState(
            entity_id=entity_id,
            winning_epoch=winner.epoch,
            winning_version=winner.version,
            winning_sha256=winner.payload_sha256,
            authoritative_payload=winner.payload,
            stale_replica_ids_for_repair=stale_ids,
        )