# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-speculative-tool-call-reconciliation-as-a-two-phase-commit-rather-than-a-compensating-rollback.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-speculative-tool-call-reconciliation-as-a-two-phase-commit-rather-than-a-compensating-rollback.warrant.json
# Markdown SHA-256:   0c4419cd6063c1e705cc6b50e4b5a895d96fa4414548342d4de236e777609ede
# Code AST SHA-256:   5b6bc6c0edf7047c26235abe17b67552e47e9fef074d543cef90d6ffafe96de4
# Verified Classes:   UnsafeCompensationError, Branch
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field


class UnsafeCompensationError(RuntimeError):
    """Raised when compensating rollback is attempted across a branch boundary
    that requires two-phase commit instead."""


@dataclass
class Branch:
    branch_id: str
    external_calls: list[str] = field(default_factory=list)
    reads_from: set[str] = field(default_factory=set)  # other branch_ids
    shared_correlation_keys: set[str] = field(default_factory=set)


def require_two_phase_commit(branches: list[Branch], selected: str) -> bool:
    """Decide whether reconciliation must use 2PC rather than compensation.

    Returns True if 2PC is required. Raises UnsafeCompensationError if the
    caller is about to compensate while a reads-from edge or shared
    correlation key makes compensation unsound.
    """
    live = [b for b in branches if b.external_calls]
    if len(live) <= 1:
        return False

    losers = [b for b in live if b.branch_id != selected]
    winner = next((b for b in live if b.branch_id == selected), None)

    for loser in losers:
        # A loser whose effects were read by any live branch cannot be
        # compensated without invalidating that branch's state.
        for other in live:
            if other.branch_id == loser.branch_id:
                continue
            if loser.branch_id in other.reads_from:
                raise UnsafeCompensationError(
                    f"branch {other.branch_id} read from {loser.branch_id}; "
                    f"compensating {loser.branch_id} would invalidate a live branch"
                )
        # A shared external correlation key (customer, tenant, account) means
        # the loser's compensation can leak into the winner's committed effects.
        if winner is not None and loser.shared_correlation_keys & winner.shared_correlation_keys:
            raise UnsafeCompensationError(
                f"branch {loser.branch_id} and winner {winner.branch_id} share "
                f"correlation keys {loser.shared_correlation_keys & winner.shared_correlation_keys}; "
                f"compensation may leak across the commit boundary"
            )

    # Any live cross-branch read, or any shared key across the whole set,
    # forces 2PC even if the specific loser checks above passed.
    all_keys: set[str] = set()
    for b in live:
        if b.reads_from & {o.branch_id for o in live if o.branch_id != b.branch_id}:
            return True
        if all_keys & b.shared_correlation_keys:
            return True
        all_keys |= b.shared_correlation_keys
    return False