# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-reject-a-speculative-branch-before-reconciliation.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-reject-a-speculative-branch-before-reconciliation.warrant.json
# Markdown SHA-256:   0dda41937a27ecc7671fa68f89bfdd43beb0ef97a8630db57e129ca12377437a
# Code AST SHA-256:   b79716c08aa8b32e16b4a4a4efecc9c1711b3014880c76d5f3ceee1816352bb3
# Verified Classes:   SpeculationBoundaryError, EffectClass, Effect, Branch
# Defensive Guards:   2 fail-closed invariants (zero external deps)

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


class SpeculationBoundaryError(RuntimeError):
    """Raised when a branch has escaped non-idempotent effects without a
    recorded compensation handle, making logical rollback unsound."""
    pass


class EffectClass(Enum):
    PURE = "pure"
    IDEMPOTENT = "idempotent"
    COMPENSABLE = "compensable"      # non-idempotent, handle recorded at effect-time
    UNRECOVERABLE = "unrecoverable"  # non-idempotent, no handle


@dataclass
class Effect:
    kind: str
    effect_class: EffectClass
    compensation_handle: Optional[str] = None


@dataclass
class Branch:
    branch_id: str
    effects: list[Effect] = field(default_factory=list)
    result: Optional[dict] = None

    def emit(self, kind: str, effect_class: EffectClass,
             compensation_handle: Optional[str] = None) -> None:
        self.effects.append(Effect(kind, effect_class, compensation_handle))


def classify_for_reconciliation(branch: Branch) -> str:
    """Decide whether a branch may be optimistically reconciled.

    Returns one of: 'reconcile', 'compensate', 'quarantine'.
    Raises SpeculationBoundaryError if the branch is ineligible yet the
    caller attempts to proceed as if it were reconcilable.
    """
    has_unrecoverable = any(
        e.effect_class is EffectClass.UNRECOVERABLE for e in branch.effects
    )
    has_compensable = any(
        e.effect_class is EffectClass.COMPENSABLE for e in branch.effects
    )

    if has_unrecoverable:
        # Non-idempotent effect with no recorded handle: must NOT be rolled
        # back logically; quarantine for supervised compensation.
        return "quarantine"
    if has_compensable:
        return "compensate"
    return "reconcile"


def reconcile(winner: Branch, losers: list[Branch]) -> Branch:
    for loser in losers:
        verdict = classify_for_reconciliation(loser)
        if verdict == "quarantine":
            raise SpeculationBoundaryError(
                f"branch {loser.branch_id!r} emitted unrecoverable external "
                f"effects; logical rollback is unsound. Quarantine and route "
                f"to supervised compensation instead of discarding."
            )
        if verdict == "compensate":
            for e in loser.effects:
                if e.effect_class is EffectClass.COMPENSABLE:
                    # Execute recorded compensation before declaring join done.
                    _run_compensation(e.compensation_handle)
        # verdict == 'reconcile': safe to discard loser's result tree.
    return winner


def _run_compensation(handle: Optional[str]) -> None:
    if not handle:
        raise SpeculationBoundaryError(
            "compensable effect missing handle at execution time"
        )
    # ... invoke recorded compensating action ...