# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-equivocation-collapse-in-bft-task-commitment-consensus-the-quorum-intersection-gap-under-partial-synchrony.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-equivocation-collapse-in-bft-task-commitment-consensus-the-quorum-intersection-gap-under-partial-synchrony.warrant.json
# Markdown SHA-256:   f243a330d2f56313a45dc184f8564b5493957545de51c19671239ac5c57c6f98
# Code AST SHA-256:   80f5290202b10362379f4badd22926f5bcc03c2b03b67e9353e3358b53dea9ae
# Verified Classes:   CommitmentMonotonicityError, CommitmentRecord, DurableCommitmentStore
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class CommitmentMonotonicityError(RuntimeError):
    """Raised when a correct agent cannot guarantee certificate monotonicity.

    This is a SAFETY halt, not a transient error. Do not retry blindly.
    """


@dataclass(frozen=True)
class CommitmentRecord:
    task_id: str
    epoch: int
    decision: str  # e.g. "commit:<assignment-hash>"


class DurableCommitmentStore:
    """Atomic store: highest-epoch record and emitted signature share one write."""

    def __init__(self, backend):
        self._backend = backend  # must support atomic batch write + fsync

    def load_highest(self, task_id: str) -> CommitmentRecord | None:
        return self._backend.get(f"commit:{task_id}")

    def persist_and_emit(self, record: CommitmentRecord) -> str:
        # Single atomic unit: if this returns, both record and signature are durable.
        sig = hashlib.sha256(
            f"{record.task_id}|{record.epoch}|{record.decision}".encode()
        ).hexdigest()
        self._backend.atomic_batch_write(
            [(f"commit:{record.task_id}", record), (f"sig:{record.task_id}:{record.epoch}", sig)],
            fsync=True,
        )
        return sig


def sign_commitment(store: DurableCommitmentStore, record: CommitmentRecord) -> str:
    prior = store.load_highest(record.task_id)

    if prior is not None and record.epoch <= prior.epoch:
        # Refuse to sign at or below a persisted epoch: this is the invariant.
        raise CommitmentMonotonicityError(
            f"refusing to sign task={record.task_id} epoch={record.epoch} "
            f"<= persisted epoch={prior.epoch} decision={prior.decision}"
        )

    if prior is not None and prior.decision != record.decision:
        # Same slot, higher epoch, different decision: a conflicting certificate
        # is being proposed. Signing would open the quorum-intersection gap.
        raise CommitmentMonotonicityError(
            f"conflicting decision for committed task={record.task_id}: "
            f"persisted={prior.decision} proposed={record.decision}"
        )

    try:
        return store.persist_and_emit(record)
    except OSError as exc:
        # Cannot guarantee durable-before-emit. Halting is the safety-correct action.
        raise CommitmentMonotonicityError(
            f"durable store unavailable for task={record.task_id} epoch={record.epoch}: {exc}"
        ) from exc


def accept_state_sync(store: DurableCommitmentStore, task_id: str, peer_record: CommitmentRecord) -> None:
    local = store.load_highest(task_id)
    if local is not None and peer_record.epoch < local.epoch:
        raise CommitmentMonotonicityError(
            f"state-sync would regress task={task_id} from epoch={local.epoch} "
            f"to peer epoch={peer_record.epoch}; refusing lagging peer as truth source"
        )