# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-reject-a-crdt-merge-that-is-locally-convergent-but-causally-stale.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-reject-a-crdt-merge-that-is-locally-convergent-but-causally-stale.warrant.json
# Markdown SHA-256:   0c0e8fa8391e99504807ecfaddb658f1d135c99994d6759dd0b12371f35c71d5
# Code AST SHA-256:   17bb14ae12d0368c70c23b97b02017cd57d63b09906aaf6b65d21ca049f7bad2
# Verified Classes:   CausalStaleMergeError, Delta, Blackboard
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet, Hashable


class CausalStaleMergeError(RuntimeError):
    """Raised when an LWW delta is locally convergent but causally stale
    relative to the local board's composite-record causal past."""


@dataclass(frozen=True)
class Delta:
    record_id: Hashable
    field_name: str
    value: object
    hlc: tuple[int, int]
    read_set: FrozenSet[Hashable]  # causal fingerprint of the writer's read


@dataclass
class Blackboard:
    # record_id -> field -> value
    fields: dict = field(default_factory=dict)
    # record_id -> set of write-tokens the local replica has observed
    causal_past: dict = field(default_factory=dict)
    # record_id -> list of quarantined deltas awaiting re-derive
    conflict_buffer: dict = field(default_factory=dict)

    def observe(self, record_id: Hashable, token: Hashable) -> None:
        self.causal_past.setdefault(record_id, set()).add(token)

    def apply(self, delta: Delta) -> None:
        record = self.fields.setdefault(delta.record_id, {})
        past = self.causal_past.get(delta.record_id, set())

        # Boundary check: composite record + stale read-set + sibling mutated.
        sibling_mutated = bool(record) and delta.field_name not in record
        read_set_valid = delta.read_set.issubset(past)

        if sibling_mutated and not read_set_valid:
            self.conflict_buffer.setdefault(delta.record_id, []).append(delta)
            raise CausalStaleMergeError(
                f"Rejecting LWW winner for {delta.record_id}.{delta.field_name}: "
                f"read_set {sorted(map(str, delta.read_set))} is not a subset of "
                f"local causal past {sorted(map(str, past))}; sibling fields "
                f"{sorted(record)} were mutated after the writer's read-cut. "
                f"Quarantined for re-derive."
            )

        # LWW resolution: accept the delta if its HLC is strictly greater.
        existing = record.get(delta.field_name)
        if existing is None or delta.hlc > existing[0]:
            record[delta.field_name] = (delta.hlc, delta.value)
            self.observe(delta.record_id, (delta.field_name, delta.hlc))

    def drain_conflicts(self, record_id: Hashable) -> list[Delta]:
        return self.conflict_buffer.pop(record_id, [])