# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-stall-at-f-1-equivocating-agents-when-the-commitment-quorum-is-computed-over-distinct-task-ids-rather-than-distinct-agents.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-stall-at-f-1-equivocating-agents-when-the-commitment-quorum-is-computed-over-distinct-task-ids-rather-than-distinct-agents.warrant.json
# Markdown SHA-256:   f4274451b212ffb767f50e5fa42925788b570fb53f551c1e4b1350e16d54c5cf
# Code AST SHA-256:   e91422d3320b69d4f200ca7af6f7e7044b29ded8d9313907f7b37118c83e3685
# Verified Classes:   QuorumDomainError, Commit, Coordinator
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Iterable


class QuorumDomainError(RuntimeError):
    """Raised when a commit quorum is tallied over the wrong domain."""


@dataclass(frozen=True)
class Commit:
    agent_id: str
    task_id: str
    signature: str


@dataclass
class Coordinator:
    n: int
    f: int
    authority_assigned: dict[str, str] = field(default_factory=dict)  # task_id -> agent_id

    @property
    def quorum(self) -> int:
        return 2 * self.f + 1

    def tally_by_task_id(self, commits: Iterable[Commit]) -> set[str]:
        # UNSAFE: counts distinct strings, not distinct signers.
        seen: set[str] = set()
        for c in commits:
            seen.add(c.task_id)
        return seen

    def tally_by_agent(self, commits: Iterable[Commit]) -> set[str]:
        # SAFE: counts distinct signers, and rejects unassigned task_ids.
        signers: dict[str, str] = {}
        for c in commits:
            assigned = self.authority_assigned.get(c.task_id)
            if assigned is not None and assigned != c.agent_id:
                raise QuorumDomainError(
                    f"agent {c.agent_id} committed task {c.task_id} "
                    f"assigned to {assigned}"
                )
            if c.agent_id in signers:
                # duplicate signer: same logical vote, cannot inflate quorum
                continue
            signers[c.agent_id] = c.task_id
        return set(signers)

    def commit(self, commits: Iterable[Commit], *, dedup: str = "agent") -> set[str]:
        commits = list(commits)
        if dedup == "task_id":
            keys = self.tally_by_task_id(commits)
            if len(keys) >= self.quorum:
                # This branch is the failure mode: a single Byzantine agent
                # can reach it alone.
                return keys
            return set()
        signers = self.tally_by_agent(commits)
        if len(signers) < self.quorum:
            return set()
        # invariant: >= f+1 honest signers iff distinct-agent count >= quorum
        if len(signers) < self.f + 1:
            raise QuorumDomainError(
                f"quorum reached with {len(signers)} distinct signers, "
                f"below honesty floor f+1={self.f + 1}"
            )
        return set(signers.values())


def demonstrate_boundary() -> None:
    coord = Coordinator(n=7, f=2)
    byz = "agent-evil"
    forged = [
        Commit(agent_id=byz, task_id=f"task-{i}", signature=f"sig-{i}")
        for i in range(coord.quorum)
    ]
    # Safe path: one signer cannot reach quorum.
    assert coord.commit(forged, dedup="agent") == set()
    # Unsafe path: same messages satisfy the quorum by task-id.
    unsafe = coord.commit(forged, dedup="task_id")
    assert len(unsafe) == coord.quorum
    raise QuorumDomainError(
        f"quorum satisfied by {len(unsafe)} task-ids from a single signer "
        f"{byz}; dedup key must be agent_id, not task_id"
    )


if __name__ == "__main__":
    demonstrate_boundary()