# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-roll-back-a-speculative-branch-in-multi-agent-execution.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-roll-back-a-speculative-branch-in-multi-agent-execution.warrant.json
# Markdown SHA-256:   0fa3174d3cf7be662d34fb590daec818f36704282a19e5580cc0e0643769629f
# Code AST SHA-256:   db12a4142d88258c08b842c046a615d24fd70e8742628fb80abbad4cf5ccf93e
# Verified Classes:   BranchState, UnrollableBranchError, Effect, Branch
# Defensive Guards:   2 fail-closed invariants (zero external deps)

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


class BranchState(Enum):
    SPECULATIVE = "speculative"
    COMMITTED = "committed"
    ROLLED_BACK = "rolled_back"
    UNROLLABLE = "unrollable"


class UnrollableBranchError(RuntimeError):
    """Raised when rollback is requested on a branch whose effects have escaped
    the speculation boundary and cannot be provably compensated."""


@dataclass(frozen=True)
class Effect:
    kind: str
    buffered_private: bool
    inverse_executable: bool
    observed_by_external: bool


@dataclass
class Branch:
    branch_id: str
    state: BranchState = BranchState.SPECULATIVE
    effects: list[Effect] = field(default_factory=list)

    def is_reversible(self) -> bool:
        for e in self.effects:
            if e.buffered_private:
                continue
            if e.inverse_executable and not e.observed_by_external:
                continue
            return False
        return True


def rollback_branch(branch: Branch) -> Branch:
    """Roll back a losing speculative branch, refusing when effects escaped."""
    if branch.state is not BranchState.COMMITTED:
        raise UnrollableBranchError(
            f"branch {branch.branch_id}: rollback requires COMMITTED state, "
            f"got {branch.state.value}"
        )
    if not branch.is_reversible():
        branch.state = BranchState.UNROLLABLE
        escaped = [e.kind for e in branch.effects
                   if not e.buffered_private
                   and not (e.inverse_executable and not e.observed_by_external)]
        raise UnrollableBranchError(
            f"branch {branch.branch_id}: cannot roll back; escaped effects: {escaped}. "
            f"Promote to winning outcome or escalate to compensating workflow."
        )
    branch.state = BranchState.ROLLED_BACK
    return branch


def arbitrate(branches: Iterable[Branch], score_fn) -> Branch:
    """Select the winning branch, but never roll back an escaped loser."""
    ranked = sorted(branches, key=score_fn, reverse=True)
    winner, losers = ranked[0], ranked[1:]
    for loser in losers:
        if not loser.is_reversible():
            # The nominal loser has escaped; the only consistent outcome is
            # to adopt it and demote the nominal winner.
            loser.state = BranchState.COMMITTED
            return loser
        rollback_branch(loser)
    winner.state = BranchState.COMMITTED
    return winner