# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-abort-a-speculative-branch-rollback-invalidation-under-byzantine-sibling-divergence.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-abort-a-speculative-branch-rollback-invalidation-under-byzantine-sibling-divergence.warrant.json
# Markdown SHA-256:   01cc237eebbfd459f2a6efeceb53f68a1d25a1866f3966be0814a2dfcb94617f
# Code AST SHA-256:   1081eb9d33320c41ce7b73dae73d2e1fa3a93ebcbf46844f98f7179020d447cb
# Verified Classes:   ByzantineRollbackViolation, Effect
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class ByzantineRollbackViolation(RuntimeError):
    """Raised when a rollback target predates a committed effect in its namespace."""
    pass


@dataclass(frozen=True)
class Effect:
    branch_id: str
    generation: int
    namespace: str


def assert_rollback_safe(
    target_generation: int,
    branch_namespace: str,
    committed_effects: list[Effect],
    rollback_set: frozenset[str],
) -> None:
    """Enforce the cross-branch rollback invariant before replaying a checkpoint.

    Safe iff every committed effect touching branch_namespace was produced at or
    after target_generation, OR its producing branch is itself being rolled back.
    """
    for effect in committed_effects:
        if effect.namespace != branch_namespace:
            continue
        if effect.generation >= target_generation:
            continue
        if effect.branch_id in rollback_set:
            # Sibling is co-rolled-back; generation ordering holds across the set.
            continue
        raise ByzantineRollbackViolation(
            f"rollback target gen={target_generation} predates committed effect "
            f"from branch={effect.branch_id} gen={effect.generation} in "
            f"namespace={branch_namespace!r}; escalate to subtree invalidation, "
            f"do not retry"
        )