# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-speculative-tool-call-branches-from-corrupting-committed-state.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-speculative-tool-call-branches-from-corrupting-committed-state.warrant.json
# Markdown SHA-256:   ebc24f485e90d17ff9695ce69639badee279a4852e59d3e75cac4288433314e3
# Code AST SHA-256:   50f5a937d09de44057cf480f4a08aeca7d96f2d7ad5a71ed2253e4b73f253077
# Verified Classes:   SpeculativeCommitError, EffectToken, Branch, CommitCoordinator
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Dict, Set


class SpeculativeCommitError(RuntimeError):
    """Raised when a speculative branch violates a commit-protocol invariant."""


@dataclass
class EffectToken:
    tool: str
    canonical_args: str
    goal_id: str

    def key(self) -> str:
        # (I2): identity is branch-independent.
        return f"{self.goal_id}:{self.tool}:{self.canonical_args}"


@dataclass
class Branch:
    name: str
    tokens: Set[str] = field(default_factory=set)
    externally_observable: Set[str] = field(default_factory=set)


class CommitCoordinator:
    IRREVERSIBLE_TOOLS = {"send_email", "charge_card", "http_post_external"}

    def __init__(self, goal_id: str) -> None:
        self.goal_id = goal_id
        self.registered: Dict[str, str] = {}   # token key -> branch name
        self.committed_branch: str | None = None

    def register(self, branch: Branch, token: EffectToken) -> None:
        # (I4): no registration after commit.
        if self.committed_branch is not None:
            raise SpeculativeCommitError(
                f"registration of {token.key()} after commit by "
                f"{self.committed_branch}; branch {branch.name} is stale"
            )
        # (I2): token collision across branches is allowed and deduplicated.
        key = token.key()
        owner = self.registered.get(key)
        if owner is None:
            self.registered[key] = branch.name
        branch.tokens.add(key)

    def dispatch(self, branch: Branch, token: EffectToken) -> None:
        # (I1): registration must precede observability.
        if token.key() not in branch.tokens:
            raise SpeculativeCommitError(
                f"branch {branch.name} dispatched {token.tool} before "
                f"registering effect token {token.key()}"
            )
        if token.tool in self.IRREVERSIBLE_TOOLS:
            # Decision boundary: irreversible tools may not run speculatively.
            raise SpeculativeCommitError(
                f"tool {token.tool} is externally observable and non-idempotent; "
                f"refusing speculative dispatch on branch {branch.name}. "
                f"Serialize to commit phase or set allow_speculative_irreversible."
            )
        branch.externally_observable.add(token.key())

    def commit(self, branch: Branch) -> None:
        if self.committed_branch is not None:
            raise SpeculativeCommitError(
                f"goal {self.goal_id} already committed by {self.committed_branch}"
            )
        self.committed_branch = branch.name

    def discard(self, branch: Branch) -> None:
        # (I3): discard must reverse or raise; never silent success.
        if branch.name == self.committed_branch:
            raise SpeculativeCommitError(
                f"cannot discard committed branch {branch.name}"
            )
        unreversible = [
            k for k in branch.tokens
            if k in branch.externally_observable and k in self.registered
            and self.registered[k] == branch.name
        ]
        if unreversible:
            raise SpeculativeCommitError(
                f"discard of branch {branch.name} cannot reverse "
                f"{len(unreversible)} effect(s): {sorted(unreversible)}"
            )
        for k in branch.tokens:
            if self.registered.get(k) == branch.name:
                del self.registered[k]