# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-cancels-a-speculative-tool-call-after-the-orchestrator-has-already-committed-the-branch-the-irreversible-side-effect-inv.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-cancels-a-speculative-tool-call-after-the-orchestrator-has-already-committed-the-branch-the-irreversible-side-effect-inv.warrant.json
# Markdown SHA-256:   8c8c3af654a1c013519cb9bd8d783957a83d88fd8f073c26f0e9ec371ee6d381
# Code AST SHA-256:   a61ceb3ee70305f629d4d2bff0d9fc962b2fdd21bb4de06dc2d52d09ff30176a
# Verified Classes:   SpeculationIntegrityError, CallState, SpeculativeCall, Branch
# Defensive Guards:   3 fail-closed invariants (zero external deps)

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


class SpeculationIntegrityError(RuntimeError):
    """Raised when a speculative branch's rollback cannot be proven effect-free."""


class CallState(Enum):
    PENDING = "pending"
    COMMITTED = "committed"
    CANCELLED = "cancelled"
    UNKNOWN = "unknown"


@dataclass
class SpeculativeCall:
    call_id: str
    callee: str
    idempotent: bool
    commit_observable: bool
    state: CallState = CallState.PENDING


@dataclass
class Branch:
    branch_id: str
    calls: list[SpeculativeCall] = field(default_factory=list)
    aborted: bool = False
    escaped_effects: list[str] = field(default_factory=list)


def request_cancel(call: SpeculativeCall) -> None:
    # Cancellation is advisory once the request has left the orchestrator.
    # The callee decides; we can only observe the terminal state.
    if call.state is CallState.PENDING:
        call.state = CallState.UNKNOWN


def rollback_branch(branch: Branch) -> None:
    if not branch.calls:
        branch.aborted = True
        return

    for call in branch.calls:
        request_cancel(call)

    # Drain: do NOT mark aborted until every call reaches a terminal state.
    # Any call that committed after the cancel is an escaped effect and must
    # be attributed to this branch, not silently dropped.
    for call in branch.calls:
        if call.state is CallState.UNKNOWN:
            raise SpeculationIntegrityError(
                f"branch {branch.branch_id}: call {call.call_id} to "
                f"{call.callee} has no terminal state; cannot prove "
                f"rollback is effect-free"
            )
        if call.state is CallState.COMMITTED:
            branch.escaped_effects.append(call.call_id)

    if branch.escaped_effects and not all(c.idempotent for c in branch.calls):
        raise SpeculationIntegrityError(
            f"branch {branch.branch_id}: aborted with escaped effects "
            f"{branch.escaped_effects}; reconcile before fallback executes "
            f"or the fallback will duplicate them"
        )

    branch.aborted = True


def dispatch_speculative(branch: Branch, call: SpeculativeCall) -> None:
    if not call.idempotent and not call.commit_observable:
        raise SpeculationIntegrityError(
            f"refusing to speculatively dispatch non-idempotent call "
            f"{call.call_id} to {call.callee} with unobservable commit point"
        )
    branch.calls.append(call)