# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a-speculative-branch-diverge-silently-when-its-rollback-log-is-truncated-by-a-concurrent-reconciliation-pass.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a-speculative-branch-diverge-silently-when-its-rollback-log-is-truncated-by-a-concurrent-reconciliation-pass.warrant.json
# Markdown SHA-256:   3a38ffa4f31438bd8201aa06e43530acefa67191d15b58a058ca1bf0596d8e63
# Code AST SHA-256:   5bd1fe5e6521c1d5d950599b936482789a732b4302817bb26120ddb64d13d316
# Verified Classes:   SilentDivergenceError, RollbackLog
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Optional


class SilentDivergenceError(RuntimeError):
    """Raised when a speculative branch's rollback replayed a log range
    that was concurrently compacted, producing an internally-consistent
    but causally-invalid resume point."""


@dataclass
class RollbackLog:
    entries: list
    compaction_events: list = field(default_factory=list)

    def seq_range(self) -> tuple:
        if not self.entries:
            return (0, 0)
        return (self.entries[0]["seq"], self.entries[-1]["seq"])

    def compact(self, up_to_seq: int) -> None:
        self.compaction_events.append((up_to_seq, self.seq_range()))
        self.entries = [e for e in self.entries if e["seq"] > up_to_seq]


def rollback_branch(branch_id: str, log: RollbackLog, pinned: bool) -> dict:
    """Replay a branch's speculative writes from the log tail.

    `pinned` indicates whether the caller holds a read-lease on the
    log's sequence range for the duration of the replay. If not pinned,
    a concurrent compaction can truncate the range mid-replay and the
    branch will silently resume from a stale baseline.
    """
    start_range = log.seq_range()
    replayed = list(reversed(log.entries))  # tail-first replay
    end_range = log.seq_range()

    if not pinned and start_range != end_range:
        raise SilentDivergenceError(
            f"branch {branch_id} replayed log range {start_range} but log "
            f"range changed to {end_range} mid-replay; concurrent compaction "
            f"events: {log.compaction_events}. Resume point is causally "
            f"invalid — abort branch and re-speculate from a pinned snapshot."
        )

    if not replayed:
        raise SilentDivergenceError(
            f"branch {branch_id} found empty rollback log; cannot establish "
            f"a valid baseline. Refusing to resume from implicit empty state."
        )

    return {"branch": branch_id, "baseline_seq": replayed[-1]["seq"]}


# --- reproduction of the observed failure ---
log = RollbackLog(entries=[{"seq": i, "write": f"w{i}"} for i in range(1, 413)])
try:
    # reconciler compacts entries 1..87 while branch B replays unpinned
    log.compact(up_to_seq=87)
    rollback_branch("B", log, pinned=False)
except SilentDivergenceError as e:
    print(f"caught: {e}")