# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-reconcile-speculative-branch-state-when-a-merge-rollback-invalidates-a-sibling-s-already-committed-side-effects.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-reconcile-speculative-branch-state-when-a-merge-rollback-invalidates-a-sibling-s-already-committed-side-effects.warrant.json
# Markdown SHA-256:   51c721adec506d9988f2287e0e8577768f23589ad126031f84648f5c9f98d7f5
# Code AST SHA-256:   322433635f6dda5c68fc10c561974834c6a7796c8cd4dace375309e3f75b253a
# Verified Classes:   BranchStatus, MergeRollbackError, SideEffect, Branch
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from enum import Enum


class BranchStatus(str, Enum):
    SPECULATING = "SPECULATING"
    MERGED = "MERGED"
    ABANDONED = "ABANDONED"


class MergeRollbackError(RuntimeError):
    """Raised when a merge rollback would tombstone a branch whose
    non-idempotent side effects were already committed and are not
    covered by a compensation handle."""


@dataclass
class SideEffect:
    tool: str
    idempotent: bool
    compensable: bool
    handle: str | None = None  # compensation handle, if any


@dataclass
class Branch:
    branch_id: str
    status: BranchStatus = BranchStatus.SPECULATING
    ledger: list[SideEffect] = field(default_factory=list)


def commit_side_effect(branch: Branch, effect: SideEffect) -> None:
    if branch.status is not BranchStatus.SPECULATING:
        raise MergeRollbackError(
            f"branch {branch.branch_id} is {branch.status}; cannot commit new effect"
        )
    if not effect.idempotent and not effect.compensable:
        # Irreversible action inside speculation: forbidden by the boundary.
        raise MergeRollbackError(
            f"tool {effect.tool!r} is non-idempotent and non-compensable; "
            f"must be deferred past merge barrier, not committed speculatively"
        )
    branch.ledger.append(effect)


def rollback_branch(branch: Branch) -> list[SideEffect]:
    """Tombstone a branch on merge rollback. Returns compensations to run.

    The ledger is preserved, never deleted, so orphaned effects remain
    reconstructable from branch records alone.
    """
    if branch.status is BranchStatus.MERGED:
        raise MergeRollbackError(
            f"branch {branch.branch_id} already merged; rollback would orphan effects"
        )
    uncompensated = [
        e for e in branch.ledger if not e.idempotent and e.handle is None
    ]
    if uncompensated:
        raise MergeRollbackError(
            f"branch {branch.branch_id} has {len(uncompensated)} non-idempotent "
            f"effect(s) with no compensation handle: "
            f"{[e.tool for e in uncompensated]}"
        )
    branch.status = BranchStatus.ABANDONED
    return [e for e in branch.ledger if e.compensable and e.handle]