# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-rollback-contamination-when-speculative-subtasks-share-a-mutable-reconciliation-ledger.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-rollback-contamination-when-speculative-subtasks-share-a-mutable-reconciliation-ledger.warrant.json
# Markdown SHA-256:   981958a3f953591473cc746d944753af326188288c7169fe0b7ed089ec78d640
# Code AST SHA-256:   a475a429852cf8e60d3b9866a1747549396c6659f96e29dfe0b888ee1c916e1e
# Verified Classes:   RollbackContaminationError
# Defensive Guards:   1 fail-closed invariants (zero external deps)

class RollbackContaminationError(RuntimeError):
    """Raised when a speculative loser's effects were observable to a winner
    before resolution, making value rollback insufficient."""


def resolve_speculation(winner_id, losers, ledger, read_audit):
    """
    Resolve a speculative race. Returns the set of subtasks safe to value-rollback.
    Raises RollbackContaminationError if any loser contaminated the window.
    """
    safe_to_rollback = set()

    for loser in losers:
        loser_writes = ledger.write_region(loser.epoch)
        contaminated_by = []

        for winner in winner_id:
            winner_reads = read_audit.reads_of(winner, since=loser.epoch)
            if winner_reads & loser_writes:
                contaminated_by.append(winner)

        if loser.advanced_monotonic_token():
            contaminated_by.append(f"<monotonic:{loser.token}>")

        if contaminated_by:
            raise RollbackContaminationError(
                f"Loser {loser.id} was observable before resolution; "
                f"winners {contaminated_by} conditioned decisions on it. "
                f"Window is poisoned; re-derive from pre-speculation baseline "
                f"epoch {loser.epoch - 1} instead of value-rollback."
            )

        safe_to_rollback.add(loser.id)

    return safe_to_rollback