# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-speculatively-invokes-a-non-idempotent-tool-before-branch-reconciliation-completes.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-speculatively-invokes-a-non-idempotent-tool-before-branch-reconciliation-completes.warrant.json
# Markdown SHA-256:   b0f44fb14ac98615f665f87959feee4fd129aa28a7aaa5610385ff349b342c75
# Code AST SHA-256:   cf5fb612759914955c5c2a3f741c842e49f0809cfcf4a2bdb6e8cb8635501cf0
# Verified Classes:   UnregisteredSideEffectError, IntentRecord, Coordinator, SpeculativeBranch
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field


class UnregisteredSideEffectError(RuntimeError):
    """Raised when a speculative branch attempts a non-idempotent tool call
    without a coordinator-issued intent record."""


@dataclass
class IntentRecord:
    invocation_key: str
    tool: str
    branch_id: str
    committed: bool = False
    compensated: bool = False


class Coordinator:
    def __init__(self) -> None:
        self._intents: dict[str, IntentRecord] = {}

    def request_effect(self, branch_id: str, tool: str, payload: dict, idempotent: bool) -> str:
        key = hashlib.sha256(f"{tool}:{sorted(payload.items())}".encode()).hexdigest()[:16]
        if key not in self._intents:
            self._intents[key] = IntentRecord(invocation_key=key, tool=tool, branch_id=branch_id)
        elif not idempotent:
            # Duplicate speculative request: suppress, do not re-emit.
            return key
        return key

    def commit_branch(self, branch_id: str) -> None:
        for rec in self._intents.values():
            if rec.branch_id == branch_id:
                rec.committed = True

    def rollback_branch(self, branch_id: str) -> list[str]:
        """Return invocation keys that must be compensated or deduplicated."""
        pending = []
        for rec in self._intents.values():
            if rec.branch_id == branch_id and not rec.committed:
                pending.append(rec.invocation_key)
        return pending


class SpeculativeBranch:
    def __init__(self, branch_id: str, coordinator: Coordinator) -> None:
        self.branch_id = branch_id
        self.coordinator = coordinator

    def invoke_tool(self, tool: str, payload: dict, idempotent: bool, registered: bool) -> str:
        if not idempotent and not registered:
            raise UnregisteredSideEffectError(
                f"branch {self.branch_id} attempted non-idempotent tool {tool!r} "
                f"without coordinator intent registration; effect would survive rollback"
            )
        return self.coordinator.request_effect(self.branch_id, tool, payload, idempotent)


def run_speculation(coordinator: Coordinator, branch: SpeculativeBranch) -> None:
    # Safe: idempotent and/or pre-registered.
    branch.invoke_tool("send_notification", {"to": "oncall"}, idempotent=False, registered=True)
    # Unsafe: this is the boundary failure.
    branch.invoke_tool("create_ticket", {"title": "phantom"}, idempotent=False, registered=False)


if __name__ == "__main__":
    c = Coordinator()
    b = SpeculativeBranch("branch-B", c)
    try:
        run_speculation(c, b)
    except UnregisteredSideEffectError as e:
        print("blocked:", e)
    print("pending compensation on rollback:", c.rollback_branch("branch-B"))