# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-divergence-between-speculative-branches-and-committed-state-in-multi-agent-rollouts.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-divergence-between-speculative-branches-and-committed-state-in-multi-agent-rollouts.warrant.json
# Markdown SHA-256:   ad27db77a0d463dc976b8e50d135025042f18d864660abb0d21fd2fc4616bf87
# Code AST SHA-256:   47f6d538536447579fc4e741a50b478da36fb49a9864ec234e5bf76439bebe2e
# Verified Classes:   SpeculationError, SpeculativeEffectViolation, ReservationExhausted, StaleReadSet, CallKind, Branch, Coordinator
# Defensive Guards:   6 fail-closed invariants (zero external deps)

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


class SpeculationError(RuntimeError):
    """Base for speculative-execution invariant violations."""


class SpeculativeEffectViolation(SpeculationError):
    """A branch attempted an irreversible EFFECT without a commit token."""


class ReservationExhausted(SpeculationError):
    """Aggregate reservations would exceed the bounded resource."""


class StaleReadSet(SpeculationError):
    """Commit CAS failed: a read version advanced or a lease was lost."""


class CallKind(Enum):
    PROBE = "probe"
    RESERVATION = "reservation"
    EFFECT = "effect"


@dataclass
class Branch:
    branch_id: str
    commit_token: Optional[str] = None
    read_set: Dict[str, int] = field(default_factory=dict)       # resource -> version
    reservation_set: Dict[str, int] = field(default_factory=dict)  # resource -> qty


class Coordinator:
    def __init__(self):
        self.versions: Dict[str, int] = {}
        self.reserved: Dict[str, int] = {}
        self.bounds: Dict[str, int] = {}
        self.leases: Dict[str, str] = {}  # resource -> branch_id

    def dispatch(self, branch: Branch, kind: CallKind, resource: str,
                 qty: int = 0, version: Optional[int] = None) -> None:
        if kind is CallKind.EFFECT:
            if branch.commit_token is None:
                raise SpeculativeEffectViolation(
                    f"branch {branch.branch_id} issued EFFECT on {resource} "
                    f"without a commit token"
                )
            return

        if kind is CallKind.PROBE:
            if version is None:
                raise SpeculationError(
                    f"PROBE on {resource} returned no version token; "
                    f"tool is misclassified"
                )
            branch.read_set[resource] = version
            return

        if kind is CallKind.RESERVATION:
            current = self.reserved.get(resource, 0)
            bound = self.bounds.get(resource, 0)
            if current + qty > bound:
                raise ReservationExhausted(
                    f"{resource}: reserved {current} + {qty} > bound {bound}"
                )
            self.reserved[resource] = current + qty
            self.leases[resource] = branch.branch_id
            branch.reservation_set[resource] = qty
            return

    def commit(self, branch: Branch) -> None:
        # Single atomic CAS over read_set and reservation_set.
        for resource, seen in branch.read_set.items():
            if self.versions.get(resource, 0) != seen:
                raise StaleReadSet(
                    f"{resource}: read version {seen} != current "
                    f"{self.versions.get(resource, 0)}"
                )
        for resource in branch.reservation_set:
            if self.leases.get(resource) != branch.branch_id:
                raise StaleReadSet(
                    f"{resource}: lease lost before commit"
                )
        branch.commit_token = f"commit-{branch.branch_id}"

    def abort(self, branch: Branch) -> None:
        # Release reservations; never compensate EFFECTs.
        for resource, qty in branch.reservation_set.items():
            if self.leases.get(resource) == branch.branch_id:
                self.reserved[resource] -= qty
                del self.leases[resource]
        branch.reservation_set.clear()
        branch.read_set.clear()


def run_branch(coord: Coordinator, branch: Branch, resource: str,
               speculative_effect: bool) -> None:
    if speculative_effect:
        # This is the failure the invariant exists to catch.
        raise SpeculativeEffectViolation(
            f"branch {branch.branch_id}: speculative EFFECT on {resource} "
            f"is forbidden; reclassify as RESERVATION"
        )
    coord.dispatch(branch, CallKind.RESERVATION, resource, qty=1)
    coord.commit(branch)