# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-consensus-gated-commit-as-invalid-when-the-speculative-tree-has-already-diverged-across-branches.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-consensus-gated-commit-as-invalid-when-the-speculative-tree-has-already-diverged-across-branches.warrant.json
# Markdown SHA-256:   c708315f46f298dab4987f37723ba0825eb3fd5ee418475dc1a4f1b673feb06a
# Code AST SHA-256:   17ea519775260d190e7b944b0997b638fb169498ff448424b1b637a1c6323192
# Verified Classes:   DivergentConsensusError, Branch
# Defensive Guards:   6 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field
from typing import Iterable


class DivergentConsensusError(RuntimeError):
    """Raised when a consensus-gated commit is attempted over non-exchangeable voters."""
    pass


@dataclass(frozen=True)
class Branch:
    branch_id: str
    parent_state_hash: str
    observation_log: tuple[str, ...]
    read_set: frozenset[str] = field(default_factory=frozenset)
    write_set: frozenset[str] = field(default_factory=frozenset)
    vote: str = ""


def _log_hash(log: Iterable[str]) -> str:
    h = hashlib.sha256()
    for entry in log:
        h.update(entry.encode("utf-8"))
        h.update(b"\x00")
    return h.hexdigest()


def validate_voters(voters: list[Branch], all_branches: list[Branch]) -> None:
    if not voters:
        raise DivergentConsensusError("empty voter set: quorum is undefined")

    # Condition 1: common parent state hash.
    parents = {b.parent_state_hash for b in voters}
    if len(parents) != 1:
        raise DivergentConsensusError(
            f"voters descend from {len(parents)} distinct parent states; "
            f"re-expand from lowest common ancestor"
        )

    # Condition 2: observation-log prefix consistency.
    logs = [b.observation_log for b in voters]
    longest = max(logs, key=len)
    for log in logs:
        if tuple(longest[: len(log)]) != log:
            raise DivergentConsensusError(
                "voter observation logs are not prefix-consistent; "
                "some voter is missing an observation used by another"
            )

    # Condition 3: no cross-branch causal contamination.
    voter_ids = {b.branch_id for b in voters}
    for other in all_branches:
        if other.branch_id in voter_ids:
            continue
        if other.write_set & frozenset().union(*(v.read_set for v in voters)):
            raise DivergentConsensusError(
                f"non-voter branch {other.branch_id} wrote state read by voters; "
                f"votes are causally entangled"
            )


def consensus_commit(voters: list[Branch], all_branches: list[Branch], quorum: int) -> str:
    validate_voters(voters, all_branches)

    # Condition 4: quorum over distinct state hashes, not raw branch count.
    distinct = {(_log_hash(b.observation_log), b.parent_state_hash) for b in voters}
    if len(distinct) < quorum:
        raise DivergentConsensusError(
            f"only {len(distinct)} distinct voter states for quorum {quorum}; "
            f"correlated votes do not constitute independent evidence"
        )

    tally: dict[str, int] = {}
    for b in voters:
        tally[b.vote] = tally.get(b.vote, 0) + 1
    winner, count = max(tally.items(), key=lambda kv: kv[1])
    if count < quorum:
        raise DivergentConsensusError(f"no action reached quorum {quorum}: {tally}")
    return winner