# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-equivocation-from-latency-in-bft-task-commitment-when-quorum-certificates-lag.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-equivocation-from-latency-in-bft-task-commitment-when-quorum-certificates-lag.warrant.json
# Markdown SHA-256:   848e3138c7842abe2c28fafbfb69c0e2a46c472298fa1786e643805ce25fcf25
# Code AST SHA-256:   73d55e199af4fdef6b0b0d68d1321fb8ba7b2ee298260092b275d2284b41ef30
# Verified Classes:   EquivocationEvidenceError, ConsensusSafetyViolation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class EquivocationEvidenceError(RuntimeError):
    """Raised when an agent attempts to act on equivocation without dual-signed proof."""
    pass


class ConsensusSafetyViolation(RuntimeError):
    """Raised when dual-signed evidence proves Byzantine equivocation at (height, view)."""
    pass


def classify_conflict(
    local_hash: str,
    observed: list[dict],
    f: int,
    elapsed_s: float,
    view_change_timeout_s: float,
) -> str:
    """
    observed: list of {"signer": str, "height": int, "view": int,
                       "hash": str, "sig_valid": bool}
    Returns one of: "commit", "wait_lag", "equivocation", "escalate_timeout".
    """
    by_signer: dict[str, set[str]] = {}
    for m in observed:
        if not m["sig_valid"]:
            continue
        by_signer.setdefault(m["signer"], set()).add(m["hash"])

    # Sharp criterion: same signer, same (h,v), distinct hashes.
    for signer, hashes in by_signer.items():
        if len(hashes) > 1:
            raise ConsensusSafetyViolation(
                f"dual-signed evidence from {signer} over {sorted(hashes)}"
            )

    quorum = 2 * f + 1
    signers_for_local = {
        m["signer"] for m in observed
        if m["sig_valid"] and m["hash"] == local_hash
    }
    if len(signers_for_local) >= quorum:
        return "commit"

    if elapsed_s >= view_change_timeout_s:
        return "escalate_timeout"

    # Below quorum, within timeout: admissible latency hypothesis.
    # Guard against inferring equivocation from observation alone.
    if len(signers_for_local) < quorum and elapsed_s < view_change_timeout_s:
        return "wait_lag"

    raise EquivocationEvidenceError(
        "unreachable classification state; refusing to infer fault from lag"
    )