# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-commits-a-speculative-branch-that-a-byzantine-peer-has-already-invalidated.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-commits-a-speculative-branch-that-a-byzantine-peer-has-already-invalidated.warrant.json
# Markdown SHA-256:   810c6cb5340d9666a7d5892975bb0241a997b9844396f00115247ab1b09cee3c
# Code AST SHA-256:   682196606526b910c5d0cf38e37d00c39c5fd938bdc6b123ea19ef30ac677cb3
# Verified Classes:   StaleEpochCommitError, CommitCertificate, ConflictLog
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field


class StaleEpochCommitError(RuntimeError):
    """Raised when a speculative branch attempts to seal a commit certificate
    at an epoch that no longer matches the conflict-resolution log.

    This is a safety violation, not a liveness one: the local certificate is
    authentic but not authoritative, and committing it risks divergence with a
    Byzantine peer's pending (sub-quorum) invalidation.
    """


@dataclass
class CommitCertificate:
    branch_id: str
    epoch: int
    signers: frozenset[str]
    sealed: bool = False


@dataclass
class ConflictLog:
    epoch: int = 0
    pending_invalidations: dict[str, frozenset[str]] = field(default_factory=dict)

    def advance(self) -> None:
        self.epoch += 1


def seal_speculative_branch(
    cert: CommitCertificate,
    log: ConflictLog,
    cached_epoch: int,
    quorum_size: int,
) -> CommitCertificate:
    """Seal a speculative branch only if its cached epoch still matches the log.

    Re-sealing at the current epoch is deliberately rejected: the pending
    invalidation that advanced the log is about this branch, so re-sealing
    races its finalization. Abort and re-dispatch instead.
    """
    if cached_epoch > log.epoch:
        raise StaleEpochCommitError(
            f"branch {cert.branch_id!r}: cached epoch {cached_epoch} is ahead of "
            f"log epoch {log.epoch}; forged certificate suspected"
        )

    if cached_epoch < log.epoch:
        pending = log.pending_invalidations.get(cert.branch_id, frozenset())
        raise StaleEpochCommitError(
            f"branch {cert.branch_id!r}: cached epoch {cached_epoch} < log epoch "
            f"{log.epoch}; pending invalidation signers={sorted(pending)}; "
            f"abort and re-dispatch, do NOT re-seal at epoch {log.epoch}"
        )

    if len(cert.signers) < quorum_size:
        raise StaleEpochCommitError(
            f"branch {cert.branch_id!r}: {len(cert.signers)} signers < quorum "
            f"{quorum_size} at epoch {log.epoch}"
        )

    cert.sealed = True
    return cert