# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-detect-irreversible-side-effects-before-committing-a-speculative-multi-agent-branch.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-detect-irreversible-side-effects-before-committing-a-speculative-multi-agent-branch.warrant.json
# Markdown SHA-256:   e13884ef9af444a43ef9477a8ee347afe289f623b29f35c909e468f6d48b6d03
# Code AST SHA-256:   45a7b00476f3f8c78b2496f0d6af9da463054528e3254d03bf1f5184e5a78073
# Verified Classes:   EffectClass, UnsafeSpeculationError, Tool, Branch
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Iterable


class EffectClass(Enum):
    PURE = "pure"
    IDEMPOTENT = "idempotent"
    COMPENSABLE = "compensable"
    IRREVERSIBLE = "irreversible"


class UnsafeSpeculationError(RuntimeError):
    """Raised when a branch closure reaches an effect that rollback cannot undo."""


@dataclass(frozen=True)
class Tool:
    name: str
    effect: EffectClass
    accepts_idempotency_key: bool = False


@dataclass
class Branch:
    branch_id: str
    tools: list[Tool] = field(default_factory=list)
    ledger: list[str] = field(default_factory=list)


def assert_speculable(branch: Branch, commit_token_held: bool) -> None:
    """Gate fan-out: a branch may be speculated only if rollback is sound."""
    for tool in branch.tools:
        if tool.effect is EffectClass.IRREVERSIBLE and not commit_token_held:
            raise UnsafeSpeculationError(
                f"branch {branch.branch_id!r} reaches irreversible tool "
                f"{tool.name!r} without a commit token; rollback would leak "
                f"an external effect that reconciliation cannot observe"
            )
        if tool.effect is EffectClass.IDEMPOTENT and not tool.accepts_idempotency_key:
            raise UnsafeSpeculationError(
                f"tool {tool.name!r} claims idempotency but accepts no "
                f"branch-scoped key; replay under a losing branch is unsafe"
            )


def reconcile(branches: Iterable[Branch], winner_id: str) -> Branch:
    """Compensate losers before the winner observes shared external state."""
    winner: Branch | None = None
    for branch in branches:
        if branch.branch_id == winner_id:
            winner = branch
            continue
        for handle in branch.ledger:
            # Compensation must complete before merge proceeds.
            _compensate(handle)
        branch.ledger.clear()
    if winner is None:
        raise UnsafeSpeculationError(f"winner {winner_id!r} not among branches")
    return winner


def _compensate(handle: str) -> None:
    # Placeholder for the registered compensation callback.
    return None