# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-a-heterogeneous-agent-bft-quorum-as-a-liability-rather-than-a-safeguard-the-correlated-failure-threshold-coll.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-a-heterogeneous-agent-bft-quorum-as-a-liability-rather-than-a-safeguard-the-correlated-failure-threshold-coll.warrant.json
# Markdown SHA-256:   b4b1fcd5d047ce820fe9e6eead7d4d2bd0347ef670a7973a199f3db162f0c5ce
# Code AST SHA-256:   2ba8d5041d6d8582a13e0100dc48fa05238310e2c360223dc6ba435f377901cc
# Verified Classes:   CorrelatedQuorumError, ValidatorVote, QuorumResult
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field


class CorrelatedQuorumError(RuntimeError):
    """Raised when a quorum's winning coalition shares load-bearing evidence,
    collapsing the effective independent vote count below the BFT threshold."""


@dataclass(frozen=True)
class ValidatorVote:
    validator_id: str
    decision: str                       # "safe" | "unsafe"
    primary_evidence: frozenset[str]    # raw artifacts actually read (diff, log, schema)
    claim_evidence: frozenset[str]      # summaries / prior messages relied upon


@dataclass
class QuorumResult:
    committed: bool
    effective_independent_votes: int
    required: int
    warning: str | None = None


def assess_quorum(
    votes: list[ValidatorVote],
    *,
    f: int,
    load_bearing_artifact: str,
) -> QuorumResult:
    """Evaluate a vote set against 3f+1 using *evidence-path* independence.

    A vote counts toward effective independence only if it does not rely on the
    load-bearing artifact as a *claim* (i.e., it read a primary source instead).
    Two votes sharing the same claim artifact are treated as one effective vote.
    """
    required = 2 * f + 1
    winning = [v for v in votes if v.decision == "safe"]

    # Group winning votes by the claim artifact they leaned on, if any.
    claim_groups: dict[str, int] = {}
    independent = 0
    for v in winning:
        leaned_on_claim = load_bearing_artifact in v.claim_evidence
        read_primary = load_bearing_artifact in v.primary_evidence
        if leaned_on_claim and not read_primary:
            # Correlated: collapse into a single effective vote per shared claim.
            key = load_bearing_artifact
            if key not in claim_groups:
                claim_groups[key] = 1
                independent += 1
            # additional votes on same claim add zero effective independence
        else:
            independent += 1

    if len(winning) >= required and independent < required:
        raise CorrelatedQuorumError(
            f"Quorum reached {len(winning)}/{len(votes)} votes but only "
            f"{independent} independent evidence paths (need {required}). "
            f"Load-bearing artifact {load_bearing_artifact!r} was shared as a "
            f"claim across {claim_groups.get(load_bearing_artifact, 0)} "
            f"collapsed vote(s). Downgrade to deterministic-check escalation."
        )

    warning = None
    if independent < len(winning):
        warning = (
            f"{len(winning) - independent} vote(s) collapsed onto shared claim "
            f"{load_bearing_artifact!r}; effective independence {independent}."
        )
    return QuorumResult(
        committed=len(winning) >= required,
        effective_independent_votes=independent,
        required=required,
        warning=warning,
    )