# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-abandon-speculative-branch-state-the-read-after-write-divergence-boundary-in-parallel-agent-branches-with-rollback-and-merge-rec.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-abandon-speculative-branch-state-the-read-after-write-divergence-boundary-in-parallel-agent-branches-with-rollback-and-merge-rec.warrant.json
# Markdown SHA-256:   23ed125e8fdbb7a9ab99cfc51bf86d654a02be4c068ba7def979ea9090ee2186
# Code AST SHA-256:   a6cbcb27a6925a1eefe22613327daf78e3cbb1714f62812cf9e0825bdd9b623e
# Verified Classes:   SpeculativeDivergenceError
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class SpeculativeDivergenceError(RuntimeError):
    """Raised when a rollback would invalidate a surviving branch's read set."""
    pass


def rollback_and_merge(
    abandoned_branch: str,
    abandoned_writes: set[str],
    survivors: dict[str, set[str]],  # branch -> read set over [fork, rollback]
    read_log_available: bool,
) -> list[str]:
    """Roll back an abandoned branch and return survivors safe to merge.

    Raises SpeculativeDivergenceError if a survivor observed the abandoned
    branch's writes and the caller has not opted into re-derivation.
    """
    if not read_log_available:
        # Conservative: cannot prove isolation, so no survivor is safe to merge.
        raise SpeculativeDivergenceError(
            f"read-set log unavailable; cannot prove rollback of {abandoned_branch!r} "
            f"is read-isolated from survivors {sorted(survivors)}; "
            f"re-derive all survivors from checkpoint"
        )

    contaminated = sorted(
        branch for branch, reads in survivors.items()
        if reads & abandoned_writes
    )
    if contaminated:
        raise SpeculativeDivergenceError(
            f"rollback of {abandoned_branch!r} would invalidate survivors "
            f"{contaminated}: their read sets intersect writes "
            f"{sorted(abandoned_writes)}; re-derive these branches, do not merge"
        )

    return sorted(survivors)