# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-of-byzantine-fault-tolerant-consensus-among-llm-agents-for-shared-state-and-task-commitment-draw-the-line-between-dissent-and-d.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-of-byzantine-fault-tolerant-consensus-among-llm-agents-for-shared-state-and-task-commitment-draw-the-line-between-dissent-and-d.warrant.json
# Markdown SHA-256:   47df3ac5e07033f30981941024779319cc6e04480f3ac03b5a107145d2269e76
# Code AST SHA-256:   8e0d07ce8009090738f212f633ff7804b6f773b8359b540377093573dba0b26f
# Verified Classes:   CorrelatedQuorumError, Vote
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from collections import defaultdict


class CorrelatedQuorumError(RuntimeError):
    """Raised when a quorum's votes are not provenance-distinct."""
    pass


@dataclass(frozen=True)
class Vote:
    agent_id: str
    commitment: str
    provenance_cluster: str  # hash of the input corpus the agent conditioned on
    signature_valid: bool


def evaluate_quorum(votes: list[Vote], f: int) -> str:
    # Only protocol-conformant votes enter the quorum.
    valid = [v for v in votes if v.signature_valid]
    if not valid:
        raise CorrelatedQuorumError("no protocol-conformant votes")

    # Group by commitment, then count DISTINCT provenance clusters per commitment.
    by_commitment: dict[str, set[str]] = defaultdict(set)
    for v in valid:
        by_commitment[v.commitment].add(v.provenance_cluster)

    required_clusters = 2 * f + 1
    for commitment, clusters in by_commitment.items():
        if len(clusters) >= required_clusters:
            return commitment

    # No commitment reached provenance-distinct supermajority.
    # Diagnose the most common failure: a large vote count that collapses
    # to a single provenance cluster (correlated-input agreement).
    top_commitment = max(by_commitment, key=lambda c: len(by_commitment[c]))
    top_clusters = by_commitment[top_commitment]
    raw_support = sum(1 for v in valid if v.commitment == top_commitment)
    if raw_support >= required_clusters and len(top_clusters) < required_clusters:
        raise CorrelatedQuorumError(
            f"commitment '{top_commitment}' has {raw_support} raw votes but only "
            f"{len(top_clusters)} provenance cluster(s); required {required_clusters}. "
            f"Agreement is an artifact of correlated input, not independent verification."
        )
    raise CorrelatedQuorumError(
        f"no commitment reached {required_clusters} provenance-distinct clusters"
    )


def commit_shared_state(votes: list[Vote], f: int) -> str:
    try:
        return evaluate_quorum(votes, f)
    except CorrelatedQuorumError as e:
        # Do NOT commit; escalate to provenance-diversified re-query.
        raise CorrelatedQuorumError(f"commit refused: {e}") from e