# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-byzantine-fault-tolerant-consensus-fail-to-commit-when-the-quorum-intersection-is-preserved-but-equivocating-replicas-split-th.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-byzantine-fault-tolerant-consensus-fail-to-commit-when-the-quorum-intersection-is-preserved-but-equivocating-replicas-split-th.warrant.json
# Markdown SHA-256:   c706ccc995ba19df1a6ed0d26d45c86fe2e47646597f7cb7ac47e2e0c5518b72
# Code AST SHA-256:   526a461b783be0a956e5ad82d2d70241c0723be86a4c587894d276a613dc65a0
# Verified Classes:   CommitmentIdentityError, CommitCertificate
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class CommitmentIdentityError(RuntimeError):
    """Raised when a commit certificate's identity projection is ambiguous."""
    pass


@dataclass(frozen=True)
class CommitCertificate:
    task_id: str
    payload: bytes
    epoch: int
    signatures: tuple  # quorum signatures over (task_id, payload, epoch)


def canonical_identity(cert: CommitCertificate) -> str:
    """Identity used for deduplication. Must NOT include epoch."""
    return hashlib.sha256(cert.task_id.encode() + b"|" + cert.payload).hexdigest()


def reconcile(certs: list[CommitCertificate]) -> CommitCertificate:
    """
    Given certificates observed by a replica for the same task, return the
    canonical one. Raises if the certificates disagree on payload (a real
    safety violation) but tolerates epoch disagreement (the equivocation
    boundary) by selecting the maximum epoch.
    """
    if not certs:
        raise CommitmentIdentityError("no certificates to reconcile")

    identities = {canonical_identity(c) for c in certs}
    if len(identities) > 1:
        # Different payloads for the same task_id: genuine safety violation.
        raise CommitmentIdentityError(
            f"conflicting payloads for task {certs[0].task_id}: {identities}"
        )

    # Same payload, possibly different epochs: pick canonical max epoch.
    canonical = max(certs, key=lambda c: c.epoch)

    # Sanity: every certificate must carry a quorum signature set.
    for c in certs:
        if len(c.signatures) < 3:
            raise CommitmentIdentityError(
                f"certificate for {c.task_id}@epoch={c.epoch} lacks quorum"
            )

    return canonical


def commit_task(certs: list[CommitCertificate], executor) -> str:
    """
    Commit a task exactly once. The executor is invoked at most once per
    canonical identity, regardless of how many epoch-variant certificates
    were observed.
    """
    canonical = reconcile(certs)
    identity = canonical_identity(canonical)
    executor.execute_once(identity, canonical.payload)
    return identity