# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-abort-a-speculative-multi-agent-branch-rollback-divergence-at-the-tool-result-commit-point.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-abort-a-speculative-multi-agent-branch-rollback-divergence-at-the-tool-result-commit-point.warrant.json
# Markdown SHA-256:   634f893a36ff52c013a8d539cea3bc26c3c668f177717dfb9de135b8b1b195ce
# Code AST SHA-256:   345ac195ae62fdafa301f1d51270da21f87218d7584144ce75a379326159dbfd
# Verified Classes:   RollbackBoundaryError, BranchState, Branch
# Defensive Guards:   2 fail-closed invariants (zero external deps)

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


class RollbackBoundaryError(RuntimeError):
    """Raised when a speculative branch cannot be safely rolled back."""
    pass


class BranchState(Enum):
    SPECULATIVE = "speculative"
    COMMITTED = "committed"      # side effect externalized
    CONSUMED = "consumed"        # another branch derived from this effect
    ROLLED_BACK = "rolled_back"
    QUARANTINED = "quarantined"


@dataclass
class Branch:
    name: str
    state: BranchState = BranchState.SPECULATIVE
    diverged: bool = False
    committed_at: Optional[float] = None   # externalization timestamp
    handles: list = field(default_factory=list)  # compensating-op addresses
    consumers: list = field(default_factory=list)


def decide(branch: Branch, divergence_time: float) -> str:
    """Decide rollback vs discard vs quarantine for one branch."""
    if not branch.diverged:
        return "promote"

    committed = (
        branch.state in (BranchState.COMMITTED, BranchState.CONSUMED)
        and branch.committed_at is not None
        and branch.committed_at <= divergence_time
    )

    if not committed:
        # Divergence seen before externalization: discard, do not roll back.
        return "discard"

    if branch.state == BranchState.CONSUMED:
        # A downstream branch derived from this effect; rollback must cascade.
        if not branch.consumers:
            raise RollbackBoundaryError(
                f"branch {branch.name} marked CONSUMED but records no consumers; "
                f"cannot order cascade rollback"
            )
        return "rollback_cascade"

    if not branch.handles:
        # Committed, divergent, but no compensating handles: rollback would lie.
        raise RollbackBoundaryError(
            f"branch {branch.name} committed at t={branch.committed_at} and diverged "
            f"at t={divergence_time}, but holds no compensating handles; "
            f"quarantine and escalate instead of issuing phantom rollback"
        )

    return "rollback"


def run_join(branches: list[Branch], divergence_time: float) -> dict:
    actions = {}
    for b in branches:
        try:
            actions[b.name] = decide(b, divergence_time)
        except RollbackBoundaryError as e:
            # Surface, do not swallow: a phantom rollback corrupts global state.
            actions[b.name] = "quarantine"
            actions.setdefault("_errors", []).append(str(e))
    return actions