# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-commit-under-partial-synchrony-the-quorum-intersection-invariant-as-a-liveness-safety-boundary-in-bft-mul.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-commit-under-partial-synchrony-the-quorum-intersection-invariant-as-a-liveness-safety-boundary-in-bft-mul.warrant.json
# Markdown SHA-256:   67e7b75139e4a8f51bf53c2e81dec1a1da049e49d03ecae4581b6c9f7a9ae975
# Code AST SHA-256:   ec88f4eb6ea3ddcd4970b0f77178d7ef555a7a102f6ad5a96e9a634f18b6692b
# Verified Classes:   QuorumIntersectionViolation, ReplicaLock
# Defensive Guards:   7 fail-closed invariants (zero external deps)

class QuorumIntersectionViolation(RuntimeError):
    """Raised when a commit certificate cannot be tied to a prior committed quorum
    via the local lock, i.e. the quorum-intersection invariant is unverifiable."""
    pass


class ReplicaLock:
    def __init__(self, replica_id: int, n: int, f: int):
        assert n >= 3 * f + 1, "quorum intersection requires n >= 3f+1"
        self.replica_id = replica_id
        self.n = n
        self.f = f
        self.locked_qc_view = -1          # highest view this replica voted for
        self.committed_view = -1
        self.faulty_signers = set()       # self-reported equivocation evidence

    def _verify_certificate(self, cert: dict) -> None:
        signers = set(cert.get("signers", []))
        if len(signers) < 2 * self.f + 1:
            raise QuorumIntersectionViolation(
                f"certificate has {len(signers)} signers, need {2*self.f+1}"
            )
        if signers & self.faulty_signers:
            raise QuorumIntersectionViolation(
                f"certificate includes known-faulty signers {signers & self.faulty_signers}"
            )

    def commit(self, proposal: dict) -> None:
        cert = proposal.get("certificate")
        if cert is None:
            raise QuorumIntersectionViolation("proposal carries no commit certificate")
        self._verify_certificate(cert)

        high_qc = proposal.get("highQC")
        if high_qc is None:
            # Omitted highQC: cannot prove intersection with prior commit.
            raise QuorumIntersectionViolation(
                "proposal omits highQC; quorum intersection with prior commit unprovable"
            )

        if high_qc["view"] < self.locked_qc_view:
            # This is the silent-violation case: cert is valid, but committing
            # would break the lock and allow two conflicting commits.
            raise QuorumIntersectionViolation(
                f"highQC view {high_qc['view']} < local lock {self.locked_qc_view}; "
                "refusing commit to preserve quorum-intersection safety"
            )

        if cert["view"] <= self.committed_view and high_qc["view"] < self.committed_view:
            raise QuorumIntersectionViolation(
                "certificate does not chain to last committed QC"
            )

        # Safe: certificate valid, highQC present, lock respected, chain intact.
        self.locked_qc_view = max(self.locked_qc_view, high_qc["view"])
        self.committed_view = cert["view"]