# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-a-speculative-branch-reconcile-before-its-parent-commits.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-a-speculative-branch-reconcile-before-its-parent-commits.warrant.json
# Markdown SHA-256:   6932da4ad98a6fd32cfc596fb79f44bbcf01879cadf5451a2cd2ba4a4b426bfd
# Code AST SHA-256:   3f8680e1906732003fed47944c33f7e96ad996ef07e7faf6f265221d0be1a34c
# Verified Classes:   StaleBranchError, Revision, Branch, ParentState
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet


class StaleBranchError(RuntimeError):
    """Raised when a speculative branch's read-set was invalidated by parent advancement."""


@dataclass(frozen=True)
class Revision:
    number: int


@dataclass
class Branch:
    base_revision: Revision
    read_set: FrozenSet[str]
    is_decision: bool
    delta: dict = field(default_factory=dict)


class ParentState:
    def __init__(self) -> None:
        self.revision = Revision(0)
        self.written_since: dict[int, set[str]] = {}

    def advance(self, keys_written: set[str]) -> None:
        self.revision = Revision(self.revision.number + 1)
        self.written_since[self.revision.number] = set(keys_written)

    def keys_written_after(self, rev: Revision) -> set[str]:
        out: set[str] = set()
        for n, keys in self.written_since.items():
            if n > rev.number:
                out |= keys
        return out


def reconcile(branch: Branch, parent: ParentState) -> None:
    """Validate a completed branch against the current parent before merging.

    Phantom-commit guard: a delta merge never detects read-write hazards on its
    own, so we must check the branch's read-set against parent writes that
    landed after the branch's base revision.
    """
    if branch.base_revision.number == parent.revision.number:
        return  # parent did not advance; reads are still valid

    invalidated = branch.read_set & parent.keys_written_after(branch.base_revision)
    if invalidated:
        raise StaleBranchError(
            f"branch based at rev {branch.base_revision.number} read keys "
            f"{sorted(invalidated)} which the parent rewrote by rev "
            f"{parent.revision.number}; re-execute against fresh state"
        )

    if branch.is_decision and branch.base_revision.number != parent.revision.number:
        # Decisions derived from any stale parent view are suspect even without
        # direct key overlap, because plans can encode transitive assumptions.
        raise StaleBranchError(
            "decision-producing branch has a stale base revision; "
            "require re-validation before merge"
        )