# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-rollback-hazards-in-speculative-multi-agent-execution.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-rollback-hazards-in-speculative-multi-agent-execution.warrant.json
# Markdown SHA-256:   8de05c3938b8033579d3463f1086e0e9b7110f25eab8f9b09ad768323b9d3d79
# Code AST SHA-256:   61fc4a8cceb1169dc9080f4e3f69a87e6b9eaf86ed91dc34aaef98d5ccd742c0
# Verified Classes:   SpeculativeRollbackError, BranchEffect
# Defensive Guards:   2 fail-closed invariants (zero external deps)

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


class BranchEffect:
    def __init__(self, branch_id: str, resource: str, kind: str, compensable: bool):
        self.branch_id = branch_id
        self.resource = resource
        self.kind = kind  # "write", "lock", "counter", "message"
        self.compensable = compensable
        self.observed_by: set[str] = set()


def rollback_branch(branch_id: str, effects: list[BranchEffect], live_branches: set[str]) -> None:
    """Attempt to roll back a losing speculative branch.

    Raises SpeculativeRollbackError if any effect has been observed by a
    live branch or lacks a compensation path.
    """
    for eff in effects:
        if eff.branch_id != branch_id:
            continue
        leaked = eff.observed_by & live_branches
        if leaked:
            raise SpeculativeRollbackError(
                f"effect {eff.kind} on {eff.resource} from branch {branch_id} "
                f"was observed by live branches {sorted(leaked)}; "
                f"rollback would cause phantom commit — quarantine instead"
            )
        if not eff.compensable:
            raise SpeculativeRollbackError(
                f"effect {eff.kind} on {eff.resource} from branch {branch_id} "
                f"has no compensation path; cannot roll back safely"
            )
    # All effects verified abortable; apply compensations in reverse order.
    for eff in reversed([e for e in effects if e.branch_id == branch_id]):
        _compensate(eff)


def _compensate(eff: BranchEffect) -> None:
    # Idempotent compensation dispatch keyed by (resource, kind).
    ...