# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-abort-a-byzantine-quorum-rather-than-wait-for-the-next-view.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-abort-a-byzantine-quorum-rather-than-wait-for-the-next-view.warrant.json
# Markdown SHA-256:   a3088b06993003b691216a5cfc62ec96d9ea329dd0fbe12837fd9b48ccd1452b
# Code AST SHA-256:   6f0bc5c5b659bf6bdbb44d0d7df8efce85cf168a499ee721ff89c2e95236d10b
# Verified Classes:   QuorumEquivocationError, Vote
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Iterable


class QuorumEquivocationError(RuntimeError):
    """Raised when a full quorum of votes arrived but cannot be reduced to agreement.

    This is a safety signal, not a liveness signal. Do not retry the same view.
    """


@dataclass(frozen=True)
class Vote:
    replica_id: str
    view: int
    proposal_hash: str
    signature: bytes


def count_agreeing(votes: Iterable[Vote]) -> dict[str, int]:
    counts: dict[str, int] = {}
    for v in votes:
        counts[v.proposal_hash] = counts.get(v.proposal_hash, 0) + 1
    return counts


def evaluate_view(votes: list[Vote], view: int, f: int) -> str:
    """Return 'commit', 'retry', or raise QuorumEquivocationError.

    f is the byzantine fault bound; quorum = 2f+1 out of 3f+1.
    """
    quorum = 2 * f + 1

    # Rule (3): distinct, validly-signed votes for THIS view only.
    seen: set[str] = set()
    valid: list[Vote] = []
    for v in votes:
        if v.view != view:
            continue
        if not v.signature:
            continue  # drop malformed; do not count toward quorum
        if v.replica_id in seen:
            continue  # double-count guard
        seen.add(v.replica_id)
        valid.append(v)

    if len(valid) < quorum:
        # Ordinary silence: timeout path is correct.
        return "retry"

    counts = count_agreeing(valid)
    if any(c >= quorum for c in counts.values()):
        return "commit"

    # Full quorum received, no 2f+1 agreement: equivocation, not slowness.
    partition = sorted(counts.items(), key=lambda kv: -kv[1])
    raise QuorumEquivocationError(
        f"view {view}: {len(valid)} valid votes, no agreeing quorum; "
        f"partition={partition}; aborting and escalating as safety violation"
    )