# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-speculative-tool-invocation-as-irreversible-rollback-reconciliation-boundaries-in-multi-agent-parallel-branch.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-speculative-tool-invocation-as-irreversible-rollback-reconciliation-boundaries-in-multi-agent-parallel-branch.warrant.json
# Markdown SHA-256:   fdd1fefe4b46b53dd5eac8222a4e5817a927dff6b43e7448863fe6653d8650f4
# Code AST SHA-256:   4abc76ee87b28ee462d4b1ded0dec0788feebdd24caf5accedce5db2b0942727
# Verified Classes:   IrreversibleSpeculationError, ToolTier
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class IrreversibleSpeculationError(RuntimeError):
    """Raised when a speculative branch invokes a non-compensable tool
    before the planner has issued a merge commit token."""
    pass


class ToolTier:
    RECONCILABLE = "reconcilable"
    ESCROWED = "escrowed"
    IRREVERSIBLE = "irreversible"


IRREVERSIBLE_TOOLS = {
    "send_confirmation_email",
    "capture_payment",
    "dispatch_physical_shipment",
}


def guard_speculative_invocation(tool_name: str, branch_id: str,
                                 merge_token: str | None) -> None:
    """Enforce the speculation boundary before a branch calls a tool.

    Reconcilable and escrowed tools may be called with no merge token.
    Irreversible tools require a committed merge token from the planner,
    because no idempotent inverse exists and sibling branches may have
    already observed the external effect.
    """
    tier = ToolTier.IRREVERSIBLE if tool_name in IRREVERSIBLE_TOOLS else ToolTier.ESCROWED

    if tier == ToolTier.IRREVERSIBLE and merge_token is None:
        raise IrreversibleSpeculationError(
            f"branch {branch_id!r} attempted optimistic invocation of "
            f"irreversible tool {tool_name!r} with no merge token; "
            f"rollback would be non-compensable and siblings may observe "
            f"a phantom effect"
        )


def rollback_reconcile(branches: list[dict], observed_effects: dict[str, list[str]]) -> list[str]:
    """Compensate rolled-back branches in reverse topological order of
    cross-branch observation; invalidate any branch that observed a
    compensated effect."""
    invalidated: list[str] = []
    rolled_back = {b["id"] for b in branches if b["status"] == "rolled_back"}

    # Reverse topological order: compensate observers before producers.
    for branch_id in reversed(list(rolled_back)):
        for observer, seen in observed_effects.items():
            if branch_id in seen and observer not in rolled_back:
                invalidated.append(observer)

    if invalidated:
        # A surviving branch carries stale state from a rolled-back effect.
        # Refuse to silently patch; surface for re-planning.
        raise IrreversibleSpeculationError(
            f"surviving branches {invalidated!r} observed effects from "
            f"rolled-back branches {sorted(rolled_back)!r}; "
            f"merge context is contaminated and must be re-planned"
        )
    return invalidated