# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-equivocation-in-a-bft-task-commitment-protocol-before-the-equivocating-agent-s-round-has-closed.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-equivocation-in-a-bft-task-commitment-protocol-before-the-equivocating-agent-s-round-has-closed.warrant.json
# Markdown SHA-256:   e3062bca99f8cab8b84eff9d87bb4ef7890f45471258f1c1468dcba4ad1aa3e6
# Code AST SHA-256:   1cd0d2cd791c4238ae84fd1d339e91b1fa91fba653a09eaaeb13f3ecad0ccf71
# Verified Classes:   EquivocationDetected, SignedProposal, EquivocationDetector, EquivocationProof
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass
from typing import Optional


class EquivocationDetected(RuntimeError):
    """Raised when a verified conflicting proposal pair is found for one (view, seq, task)."""
    pass


@dataclass(frozen=True)
class SignedProposal:
    view: int
    seq: int
    task_id: str
    assignment_root: str
    proposer_pubkey: str
    signature: bytes
    payload: dict

    def canonical_bytes(self) -> bytes:
        # Canonical form is what the signature covers. Re-serialization must be
        # deterministic or two encodings of one assignment produce false conflicts.
        canonical = json.dumps(
            {
                "view": self.view,
                "seq": self.seq,
                "task_id": self.task_id,
                "assignment_root": self.assignment_root,
            },
            sort_keys=True,
            separators=(",", ":"),
        )
        return canonical.encode("utf-8")


def _verify(sig: SignedProposal, trusted_keys: dict) -> bool:
    # Placeholder for real Ed25519 verification against the proposer key bound to this view.
    key = trusted_keys.get(sig.proposer_pubkey)
    if key is None:
        return False
    expected = hashlib.sha256(sig.canonical_bytes() + key).digest()
    return sig.signature == expected


class EquivocationDetector:
    def __init__(self, trusted_keys: dict, buffer_limit: int = 256):
        self.trusted_keys = trusted_keys
        self.buffer: dict[tuple[int, int, str], SignedProposal] = {}
        self.buffer_limit = buffer_limit

    def observe(self, incoming: SignedProposal) -> Optional[tuple[SignedProposal, SignedProposal]]:
        if not _verify(incoming, self.trusted_keys):
            # Unverifiable proposals are never evidence. Drop and continue.
            return None

        key = (incoming.view, incoming.seq, incoming.task_id)
        prior = self.buffer.get(key)

        if prior is None:
            if len(self.buffer) >= self.buffer_limit:
                oldest = min(self.buffer, key=lambda k: (self.buffer[k].view, self.buffer[k].seq))
                del self.buffer[oldest]
            self.buffer[key] = incoming
            return None

        if prior.assignment_root == incoming.assignment_root:
            return None  # idempotent duplicate

        # Conflicting verified pair. Both must verify and share the canonical tuple.
        if not _verify(prior, self.trusted_keys):
            return None
        return (prior, incoming)

    def accuse(self, pair: tuple[SignedProposal, SignedProposal]) -> "EquivocationProof":
        a, b = pair
        if (a.view, a.seq, a.task_id) != (b.view, b.seq, b.task_id):
            raise EquivocationDetected(
                f"cannot accuse across distinct tuples: {(a.view, a.seq, a.task_id)} vs "
                f"{(b.view, b.seq, b.task_id)}"
            )
        if a.assignment_root == b.assignment_root:
            raise EquivocationDetected(
                f"cannot accuse on identical roots {a.assignment_root!r} for "
                f"(view={a.view}, seq={a.seq}, task={a.task_id})"
            )
        return EquivocationProof(a, b)


@dataclass(frozen=True)
class EquivocationProof:
    first: SignedProposal
    second: SignedProposal

    def to_wire(self) -> dict:
        return {
            "type": "EquivocationProof",
            "first": self.first.canonical_bytes().decode(),
            "first_sig": self.first.signature.hex(),
            "second": self.second.canonical_bytes().decode(),
            "second_sig": self.second.signature.hex(),
        }


def handle_proposal(detector: EquivocationDetector, incoming: SignedProposal) -> str:
    pair = detector.observe(incoming)
    if pair is None:
        return "buffered_or_idempotent"
    proof = detector.accuse(pair)
    # Broadcast proof; enter FAULT_DECLARED for (view, seq, task).
    return f"FAULT_DECLARED:{proof.first.view}:{proof.first.seq}:{proof.first.task_id}"