# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-speculative-agent-branches-from-corrupting-shared-tool-state-during-rollback.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-speculative-agent-branches-from-corrupting-shared-tool-state-during-rollback.warrant.json
# Markdown SHA-256:   40249552330dd4ffdbaa489996c615a16feffa1d6a1d453759d8fbe51196577b
# Code AST SHA-256:   9ac8f2122ff6970fa135e2f701de730c1cabe639b024d4d26085741af22ba59a
# Verified Classes:   SpeculativeRollbackError, ToolEffect, PipelineLedger
# Defensive Guards:   4 fail-closed invariants (zero external deps)

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


class SpeculativeRollbackError(RuntimeError):
    """Raised when a rolled-back branch left uncompensated shared state."""
    pass


@dataclass
class ToolEffect:
    branch_id: str
    step_index: int
    idempotency_key: str
    compensable: bool
    compensated: bool = False


@dataclass
class PipelineLedger:
    effects: List[ToolEffect] = field(default_factory=list)

    def register(self, effect: ToolEffect) -> None:
        if effect.idempotency_key.count(":") < 2:
            raise SpeculativeRollbackError(
                f"idempotency key {effect.idempotency_key!r} is not branch-scoped"
            )
        self.effects.append(effect)

    def rollback_branch(self, branch_id: str) -> None:
        aborted = [e for e in self.effects if e.branch_id == branch_id]
        for effect in reversed(aborted):
            if not effect.compensable:
                raise SpeculativeRollbackError(
                    f"branch {branch_id} preempted non-compensable effect "
                    f"at step {effect.step_index}; cannot roll back"
                )
            effect.compensated = True
        if any(not e.compensated for e in aborted):
            raise SpeculativeRollbackError(
                f"partial rollback for branch {branch_id}"
            )


def reconcile(ledger: PipelineLedger, winner: str, losers: List[str]) -> None:
    for loser in losers:
        ledger.rollback_branch(loser)
    # Verify no losing-branch effect shares a dedup key with the winner.
    winner_keys: Dict[str, str] = {
        e.idempotency_key: e.branch_id
        for e in ledger.effects
        if e.branch_id == winner
    }
    for effect in ledger.effects:
        if effect.branch_id in losers:
            continue
        owner = winner_keys.get(effect.idempotency_key)
        if owner is not None and owner != effect.branch_id:
            raise SpeculativeRollbackError(
                f"dedup collision on {effect.idempotency_key!r} between "
                f"{owner} and {effect.branch_id}"
            )