# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-abort-speculative-multi-agent-branches-before-rollback-completes-the-side-effecting-tool-call-invariant-in-branch-a.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-abort-speculative-multi-agent-branches-before-rollback-completes-the-side-effecting-tool-call-invariant-in-branch-a.warrant.json
# Markdown SHA-256:   3895bf5529b4560c7aed227c02519418de3177b6ade54f0c6a919c2ff7347025
# Code AST SHA-256:   1d2be8394bb4f770c46b3dcc296305fa15473935250f570e3c89965a3af27376
# Verified Classes:   SpeculativeRollbackInvariantError
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class SpeculativeRollbackInvariantError(RuntimeError):
    """Raised when a branch is rolled back after crossing its effect commit point."""


EFFECT_COMMITTED = "COMMITTED"
EFFECT_UNKNOWN = "UNKNOWN"
EFFECT_REJECTED = "REJECTED"


def rollback_branch(branch, effect_status: str) -> None:
    """
    Roll back a speculative branch. Only legal while the branch has not
    durably committed an external effect. COMMITTED and UNKNOWN both block
    rollback: UNKNOWN is treated as committed because an ambiguous effect
    may already be durable.
    """
    if effect_status in (EFFECT_COMMITTED, EFFECT_UNKNOWN):
        # Do not mutate branch state to ROLLED_BACK. Route to compensation.
        raise SpeculativeRollbackInvariantError(
            f"branch {branch.id!r} cannot be rolled back: "
            f"effect_status={effect_status!r}; route to COMPENSATION_REQUIRED"
        )

    if effect_status != EFFECT_REJECTED:
        raise SpeculativeRollbackInvariantError(
            f"branch {branch.id!r} has unverified effect_status={effect_status!r}; "
            "refusing rollback without commit-negative acknowledgment"
        )

    branch.state = "ROLLED_BACK"
    branch.effect_status = EFFECT_REJECTED


def select_and_rollback(branches, winner_id: str) -> None:
    for branch in branches:
        if branch.id == winner_id:
            branch.state = "COMMITTED"
            continue
        rollback_branch(branch, branch.effect_status)