# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-side-effect-duplication-in-speculative-multi-agent-branch-and-merge-execution.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-side-effect-duplication-in-speculative-multi-agent-branch-and-merge-execution.warrant.json
# Markdown SHA-256:   ee82ec83cbce9cbda95a992700b282281c5243dbd1d9efde965075f108ab807c
# Code AST SHA-256:   77b9b95f242dd0bdf1b77b33047765a12358dedfe4f5c1675f2dd18f87e41adb
# Verified Classes:   SpeculationInvariantError, EffectClass, EffectSpec, Branch, SpeculationGuard
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from __future__ import annotations

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


class SpeculationInvariantError(RuntimeError):
    """Raised when a branch attempts an effect that cannot be safely rolled back."""


class EffectClass(Enum):
    DEFERRED = "deferred"        # runs only after merge commits
    IDEMPOTENT = "idempotent"    # safe to repeat under a merge-scoped key
    COMPENSABLE = "compensable"  # has an idempotent registered undo


@dataclass(frozen=True)
class EffectSpec:
    logical_effect_id: str
    effect_class: EffectClass
    compensator: Optional[Callable[[str], None]] = None


@dataclass
class Branch:
    branch_id: str
    merge_unit_id: str
    is_speculative: bool = True
    emitted_keys: Dict[str, str] = field(default_factory=dict)  # logical_id -> merge-scoped key


class SpeculationGuard:
    def __init__(self, merge_unit_id: str) -> None:
        self.merge_unit_id = merge_unit_id
        self._committed = False
        self._effects: Dict[str, EffectSpec] = {}

    def register_effect(self, spec: EffectSpec) -> None:
        if spec.effect_class is EffectClass.COMPENSABLE and spec.compensator is None:
            raise SpeculationInvariantError(
                f"effect {spec.logical_effect_id!r} declared compensable with no compensator"
            )
        self._effects[spec.logical_effect_id] = spec

    def key_for(self, branch: Branch, logical_effect_id: str) -> str:
        # Merge-scoped key: identical across branches that intend the same effect.
        return f"{self.merge_unit_id}:{logical_effect_id}"

    def admit_effect(self, branch: Branch, logical_effect_id: str) -> str:
        spec = self._effects.get(logical_effect_id)
        if spec is None:
            raise SpeculationInvariantError(
                f"unregistered effect {logical_effect_id!r} reached from branch {branch.branch_id!r}"
            )
        if not branch.is_speculative:
            return self.key_for(branch, logical_effect_id)
        if spec.effect_class is EffectClass.DEFERRED:
            raise SpeculationInvariantError(
                f"branch {branch.branch_id!r} attempted deferred effect "
                f"{logical_effect_id!r} before merge commit"
            )
        if spec.effect_class is EffectClass.COMPENSABLE and not self._committed:
            # Allowed only if a compensator is present; caller must register it.
            if spec.compensator is None:
                raise SpeculationInvariantError(
                    f"compensable effect {logical_effect_id!r} lacks an undo action"
                )
        key = self.key_for(branch, logical_effect_id)
        branch.emitted_keys[logical_effect_id] = key
        return key

    def rollback_branch(self, branch: Branch) -> None:
        for logical_id in branch.emitted_keys:
            spec = self._effects[logical_id]
            if spec.effect_class is EffectClass.COMPENSABLE:
                assert spec.compensator is not None
                spec.compensator(self.key_for(branch, logical_id))
            elif spec.effect_class is EffectClass.IDEMPOTENT:
                # No compensation needed; dedup key handles replay.
                continue
            else:
                raise SpeculationInvariantError(
                    f"cannot roll back deferred effect {logical_id!r} from "
                    f"speculative branch {branch.branch_id!r}"
                )
        branch.emitted_keys.clear()

    def commit(self) -> None:
        self._committed = True