# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-to-reconcile-a-speculative-branch-the-rollback-watermark-invariant-in-multi-agent-speculative-execution.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-to-reconcile-a-speculative-branch-the-rollback-watermark-invariant-in-multi-agent-speculative-execution.warrant.json
# Markdown SHA-256:   e231300a5eb1727db8cbdb4a1da22901b1f8294848764999a0430e426c9a6822
# Code AST SHA-256:   62fbafcb4b04bef985e9f043188d5075ba79feed02d26a0ea9c96cdf119e7601
# Verified Classes:   UnreconcilableBranchError, Branch
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field


class UnreconcilableBranchError(RuntimeError):
    """Raised when a speculative branch's effects have escaped the rollback watermark."""

    def __init__(self, branch_id: str, watermark: int, max_effect: int, barrier: int | None):
        self.branch_id = branch_id
        self.watermark = watermark
        self.max_effect = max_effect
        self.barrier = barrier
        super().__init__(
            f"branch {branch_id} unreconcilable: watermark={watermark} "
            f"max_effect={max_effect} barrier={barrier}"
        )


@dataclass
class Branch:
    branch_id: str
    effects: list[int] = field(default_factory=list)  # effect indices, ascending
    barriers: set[int] = field(default_factory=set)   # irreversible effect indices


def reconcile(branch: Branch, watermark: int) -> str:
    """Reconcile a speculative branch only if the watermark covers all its effects."""
    if not branch.effects:
        return "reconciled:noop"

    max_effect = max(branch.effects)
    crossed_barriers = [i for i in branch.barriers if i > watermark]

    if crossed_barriers:
        raise UnreconcilableBranchError(
            branch.branch_id, watermark, max_effect, min(crossed_barriers)
        )

    if watermark < max_effect:
        # Non-barrier effects beyond the watermark may be compensated, but only
        # if the compensation is confirmed to advance the watermark.
        raise UnreconcilableBranchError(branch.branch_id, watermark, max_effect, None)

    return f"reconciled:{branch.branch_id}"


# Example: executor branch charged at index 41, failed at 58, watermark stuck at 12.
branch = Branch(
    branch_id="exec-7f3a",
    effects=[15, 22, 41, 58],
    barriers={41},  # settled charge is irreversible
)
try:
    reconcile(branch, watermark=12)
except UnreconcilableBranchError as err:
    # Quarantine, escalate to the supervisor owning the external effect.
    print(f"QUARANTINE {err.branch_id}: barrier at {err.barrier}")