# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-rollback-storms-in-speculative-multi-agent-execution-with-tool-call-fencing.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-rollback-storms-in-speculative-multi-agent-execution-with-tool-call-fencing.warrant.json
# Markdown SHA-256:   dc69350e50a7b87788658b0b346683e19fbd8c2422e42c03977501affbea1a1f
# Code AST SHA-256:   7a7e513872968b9aaf7c6bbb18042592a7c508b6a5e3b2ea7be2859e4eb577c7
# Verified Classes:   FenceClosureError, Effect, ClosureAwareFence
# Defensive Guards:   3 fail-closed invariants (zero external deps)

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


class FenceClosureError(RuntimeError):
    """Raised when a fence is committed while its dependency closure is not fully committed."""
    pass


@dataclass
class Effect:
    id: str
    kind: str                      # "write" | "read" | "deploy" | ...
    observes: Set[str] = field(default_factory=set)   # effect ids this read depended on
    fenced: bool = False
    committed: bool = False
    aborted: bool = False


class ClosureAwareFence:
    def __init__(self) -> None:
        self.effects: Dict[str, Effect] = {}

    def register(self, e: Effect) -> None:
        self.effects[e.id] = e

    def closure(self, root_id: str) -> Set[str]:
        seen: Set[str] = set()
        stack = [root_id]
        while stack:
            cur = stack.pop()
            if cur in seen:
                continue
            seen.add(cur)
            stack.extend(self.effects[cur].observes)
        return seen

    def commit(self, root_id: str) -> None:
        root = self.effects[root_id]
        if not root.fenced:
            raise FenceClosureError(f"effect {root_id} has no fence token")

        for dep_id in self.closure(root_id):
            dep = self.effects[dep_id]
            if dep.aborted:
                raise FenceClosureError(
                    f"fence {root_id} depends on aborted effect {dep_id}; "
                    f"must cascade abort, not commit"
                )
            if not dep.committed:
                raise FenceClosureError(
                    f"fence {root_id} depends on uncommitted effect {dep_id}; "
                    f"promote closure to atomic fence or wait"
                )
        root.committed = True


def reconcile(fence: ClosureAwareFence, candidate_fence_id: str) -> None:
    # The bug this guards against: checking only fence.fenced, not closure state.
    commit = fence.effects[candidate_fence_id]
    if commit.fenced and not commit.committed:
        fence.commit(candidate_fence_id)