# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a-task-commitment-as-revocable-rather-than-durable-in-byzantine-multi-agent-consensus.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a-task-commitment-as-revocable-rather-than-durable-in-byzantine-multi-agent-consensus.warrant.json
# Markdown SHA-256:   94e8f1e9df9875ea9969ba61d0ce4ca80e33412f83cd44958ab7260f80cdde13
# Code AST SHA-256:   ea0072fb8463bd1fb926bf8ee2d5bf67133a22f39eac0786dae39025c3f4d876
# Verified Classes:   RevocableCommitmentError, Commitment
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet, Optional


class RevocableCommitmentError(RuntimeError):
    """Raised when an irreversible action is attempted on a non-durable commitment."""
    pass


@dataclass(frozen=True)
class Commitment:
    cid: str
    payload: dict
    certifying_quorum: FrozenSet[str]
    view: int
    fault_set_at_certification: FrozenSet[str] = field(default_factory=frozenset)


def is_durable(
    c: Commitment,
    conflicting: list[Commitment],
    f: int,
    proven_byzantine: FrozenSet[str],
) -> bool:
    # 1. Quorum intersection against every conflicting commitment.
    for other in conflicting:
        if other.cid == c.cid:
            continue
        overlap = c.certifying_quorum & other.certifying_quorum
        if len(overlap) < 2 * f + 1:
            return False

    # 2. Fault-set stationarity: if any certifying voter is now proven faulty,
    #    the effective correct quorum may be below threshold.
    if c.certifying_quorum & proven_byzantine:
        effective = c.certifying_quorum - proven_byzantine
        if len(effective) < 2 * f + 1:
            return False

    # 3. Payload must have been semantically validated at certification time.
    if not c.payload.get("_validated", False):
        return False

    return True


def execute_irreversible_action(
    c: Commitment,
    conflicting: list[Commitment],
    f: int,
    proven_byzantine: FrozenSet[str],
    action: callable,
) -> Optional[object]:
    if not is_durable(c, conflicting, f, proven_byzantine):
        raise RevocableCommitmentError(
            f"commitment {c.cid} is revocable (view={c.view}, "
            f"quorum={sorted(c.certifying_quorum)}); refusing irreversible action. "
            f"Reserve a rollback path or wait for grace-period promotion."
        )
    return action(c.payload)