# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-of-a-speculative-branch-be-treated-as-poisoned-rather-than-rewound-when-rollback-is-interleaved-with-optimistic-task-parallelis.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-of-a-speculative-branch-be-treated-as-poisoned-rather-than-rewound-when-rollback-is-interleaved-with-optimistic-task-parallelis.warrant.json
# Markdown SHA-256:   665fafb6bc1b706e5a36293fc11dad79111189bf65298a1132ed04a9e4ed128e
# Code AST SHA-256:   9e42346ac8458a3e5f244ba4852980044496fb3ae7e62e41e9e6b089995fef55
# Verified Classes:   PoisonedBranchError
# Defensive Guards:   1 fail-closed invariants (zero external deps)

class PoisonedBranchError(RuntimeError):
    """Raised when a speculative branch's escaped effects were observed by a
    surviving sibling and therefore cannot be safely rewound."""
    pass


def rollback_branch(branch, siblings, shared_substrate):
    escaped = branch.external_effects(shared_substrate)  # writes/leases/counters
    if not escaped:
        branch.restore_checkpoint()
        return "rewound"

    observers = [
        s for s in siblings
        if s.read_provenance() & branch.effect_ids()
    ]

    if observers:
        # Cannot rewind: survivors hold premises derived from this branch.
        for s in observers:
            s.mark_non_mergeable()
        raise PoisonedBranchError(
            f"branch {branch.id} poisoned: {len(escaped)} escaped effects "
            f"observed by siblings {[s.id for s in observers]}; "
            f"escalate to invalidation fan-out and re-execute from prior checkpoint"
        )

    # No observers: escaped effects are invertible, safe to rewind.
    branch.restore_checkpoint()
    return "rewound"