# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-speculative-branch-reconciliation-as-a-rollback-barrier-rather-than-a-merge.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-speculative-branch-reconciliation-as-a-rollback-barrier-rather-than-a-merge.warrant.json
# Markdown SHA-256:   b28278a5aa2937966e65e7d92b443d91dfcbcf99f2f0011164f7b9a7f74f4652
# Code AST SHA-256:   6b52415f2b982164bbfd037a870b5fceb254e8fb6ed739d421a176f0ff4e7155
# Verified Classes:   SpeculativeReconciliationError, Branch
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Dict, Set, List


class SpeculativeReconciliationError(RuntimeError):
    """Raised when a merge is attempted across a load-bearing speculative read edge."""
    pass


@dataclass
class Branch:
    branch_id: str
    speculative_writes: Set[str] = field(default_factory=set)   # resource ids written speculatively
    speculative_reads: Dict[str, str] = field(default_factory=dict)  # resource id -> writer branch id observed
    load_bearing_reads: Set[str] = field(default_factory=set)   # subset of reads whose value is used
    committed: bool = False


def raw_s_closure(rolled_back: List[Branch], survivors: List[Branch]) -> Set[str]:
    """Transitive closure of read-after-speculative-write edges from rolled-back writers."""
    rolled_ids = {b.branch_id for b in rolled_back}
    written = set()
    for b in rolled_back:
        written |= b.speculative_writes

    reachable: Set[str] = set()
    frontier = [b for b in survivors
                if any(r in written and b.speculative_reads.get(r) in rolled_ids
                       for r in b.load_bearing_reads)]
    while frontier:
        b = frontier.pop()
        if b.branch_id in reachable:
            continue
        reachable.add(b.branch_id)
        # propagate: this branch's speculative writes may entangle further survivors
        frontier.extend(
            s for s in survivors
            if s.branch_id not in reachable
            and any(r in b.speculative_writes
                    and s.speculative_reads.get(r) == b.branch_id
                    for r in s.load_bearing_reads)
        )
    return reachable


def reconcile(rolled_back: List[Branch], survivors: List[Branch]) -> str:
    entangled = raw_s_closure(rolled_back, survivors)
    if entangled:
        raise SpeculativeReconciliationError(
            f"Merge unsafe: load-bearing speculative reads entangle survivors "
            f"{sorted(entangled)} with rolled-back writers "
            f"{[b.branch_id for b in rolled_back]}; barrier re-speculation required."
        )
    return "merge"