# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-state-corruption-during-branch-and-verify-rollback-of-speculative-multi-agent-plans.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-state-corruption-during-branch-and-verify-rollback-of-speculative-multi-agent-plans.warrant.json
# Markdown SHA-256:   45c09fc1b2bd97755811efc7b40e541b0c8e34585fe1b80f8e5e6b3cf7020b73
# Code AST SHA-256:   cd6b2e210e9a7ea94cc75409785faef060f2c4aae8f55dcef180be5b3b82e365
# Verified Classes:   SpeculativeRollbackViolation, EffectTier, EffectLedger
# Defensive Guards:   5 fail-closed invariants (zero external deps)

class SpeculativeRollbackViolation(RuntimeError):
    """Raised when a speculative branch's effects cannot be soundly rolled back
    or when a sibling observed state that rollback will invalidate."""
    pass


class EffectTier:
    ISOLATABLE = "isolatable"
    COMPENSABLE = "compensable"
    IRREVERSIBLE = "irreversible"


class EffectLedger:
    def __init__(self):
        # branch_id -> list of (effect_id, tier, compensation_callable)
        self._entries = {}
        # resource_id -> branch_id currently fencing reads on it
        self._fences = {}

    def register(self, branch_id, effect_id, tier, compensation=None):
        if tier == EffectTier.IRREVERSIBLE:
            raise SpeculativeRollbackViolation(
                f"branch {branch_id} attempted speculative irreversible effect "
                f"{effect_id}; must be deferred to commit"
            )
        if tier == EffectTier.COMPENSABLE and compensation is None:
            raise SpeculativeRollbackViolation(
                f"branch {branch_id} issued compensable effect {effect_id} "
                f"without a registered inverse; rollback would be unsound"
            )
        self._entries.setdefault(branch_id, []).append(
            (effect_id, tier, compensation)
        )

    def fence_read(self, resource_id, reader_branch, writer_branch):
        holder = self._fences.get(resource_id)
        if holder is not None and holder != reader_branch:
            raise SpeculativeRollbackViolation(
                f"branch {reader_branch} read resource {resource_id} while "
                f"branch {holder} held an unresolved speculative effect on it; "
                f"read would be invalidated by rollback"
            )
        self._fences[resource_id] = writer_branch

    def rollback(self, branch_id):
        entries = self._entries.pop(branch_id, [])
        # reverse order compensation
        for effect_id, tier, compensation in reversed(entries):
            if tier == EffectTier.COMPENSABLE:
                try:
                    compensation()
                except Exception as exc:
                    raise SpeculativeRollbackViolation(
                        f"compensation for {effect_id} on branch {branch_id} "
                        f"failed: {exc}; state is now divergent and must be "
                        f"quarantined"
                    ) from exc
        # release fences held by this branch
        for resource_id, holder in list(self._fences.items()):
            if holder == branch_id:
                del self._fences[resource_id]

    def assert_clean(self, branch_id):
        if self._entries.get(branch_id):
            raise SpeculativeRollbackViolation(
                f"branch {branch_id} committed with {len(self._entries[branch_id])} "
                f"unresolved speculative effects"
            )


def verify_and_commit(ledger, winning_branch, loser_branches):
    for loser in loser_branches:
        ledger.rollback(loser)
    ledger.assert_clean(winning_branch)
    return True