# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-equivocating-agents-from-slow-agents-at-a-commit-boundary-in-bft-task-commitment.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-equivocating-agents-from-slow-agents-at-a-commit-boundary-in-bft-task-commitment.warrant.json
# Markdown SHA-256:   6ba37d23248f5bb200831cee30511ebe57744cde7a2e695a86617d89db5e9ad2
# Code AST SHA-256:   d77914b0a20b6b63d6d9dd6616325e7f7034b6c07ad83b40da83bc27e54ed4fa
# Verified Classes:   EquivocationDetected, CommitThresholdUnreachable, Prepare
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass


class EquivocationDetected(RuntimeError):
    """Raised when a collector observes two conflicting signed payloads for one (view, task_id)."""
    pass


class CommitThresholdUnreachable(RuntimeError):
    """Raised when the view timer expires with a single payload hash and sub-threshold signatures."""
    pass


@dataclass(frozen=True)
class Prepare:
    view: int
    task_id: str
    payload_hash: str
    agent_id: str
    signature: bytes


def collect_quorum(
    prepares: list[Prepare],
    view: int,
    task_id: str,
    threshold: int,
    view_timer_expired: bool,
) -> str:
    """Return the committed payload_hash, or raise on equivocation / unreachable threshold.

    Invariant: commitment is sound only if every signature in the quorum references
    the same payload_hash under the same (view, task_id).
    """
    relevant = [p for p in prepares if p.view == view and p.task_id == task_id]
    by_hash: dict[str, list[Prepare]] = {}
    for p in relevant:
        by_hash.setdefault(p.payload_hash, []).append(p)

    if len(by_hash) > 1:
        # Dispositive evidence of equivocation, regardless of timing.
        offenders = sorted({p.agent_id for group in by_hash.values() for p in group})
        raise EquivocationDetected(
            f"conflicting payload hashes for view={view} task={task_id}: "
            f"{sorted(by_hash)}; signers={offenders}"
        )

    if not by_hash:
        if view_timer_expired:
            raise CommitThresholdUnreachable(
                f"no prepares for view={view} task={task_id} after timer expiry"
            )
        raise CommitThresholdUnreachable(
            f"no prepares yet for view={view} task={task_id}; wait"
        )

    (payload_hash, group), = by_hash.items()
    distinct_signers = {p.agent_id for p in group}
    if len(distinct_signers) >= threshold:
        return payload_hash

    if view_timer_expired:
        raise CommitThresholdUnreachable(
            f"single payload {payload_hash} but only {len(distinct_signers)}/{threshold} "
            f"signers for view={view} task={task_id}"
        )
    # Slow-agent branch: single hash, timer not fired, keep waiting.
    return None