# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-reconcile-non-idempotent-side-effects-across-speculative-multi-agent-branches-when-a-rollback-checkpoint-is-taken-after-a-tool.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-reconcile-non-idempotent-side-effects-across-speculative-multi-agent-branches-when-a-rollback-checkpoint-is-taken-after-a-tool.warrant.json
# Markdown SHA-256:   d159e32ba6f6f90a88123721e01a5d34fb48e4656b3379132400c7d4953975b1
# Code AST SHA-256:   1cc006afb76ea660528aa9b2ec38c90f6a0b4037d66fe95fac8c7ac4c6a050a9
# Verified Classes:   SpeculativeRollbackViolation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class SpeculativeRollbackViolation(RuntimeError):
    """Raised when a prune would discard a branch whose tool call already committed
    an effect that cannot be compensated or deduplicated across sibling branches."""

    def __init__(self, branch_id: str, tool: str, effect_class: str, op_key: str):
        self.branch_id = branch_id
        self.tool = tool
        self.effect_class = effect_class
        self.op_key = op_key
        super().__init__(
            f"cannot prune branch {branch_id!r}: tool {tool!r} committed "
            f"effect_class={effect_class!r} under logical op {op_key!r}; "
            f"reconcile before pruning"
        )


EFFECT_COMPENSABLE = "compensable"
EFFECT_DEDUPLICABLE = "deduplicable"
EFFECT_NEITHER = "neither"


def prune_branch(branch, registry, ledger):
    """Prune a speculative branch, refusing when a committed effect blocks rollback."""
    for call in branch.committed_calls:
        effect_class = registry.classify(call.tool)
        # A committed effect is only safely discardable if it can be reversed or
        # deduplicated by a key shared across siblings (minted at fork, not branch).
        if effect_class == EFFECT_NEITHER:
            raise SpeculativeRollbackViolation(
                branch_id=branch.id,
                tool=call.tool,
                effect_class=effect_class,
                op_key=call.logical_op_key,
            )
        if effect_class == EFFECT_DEDUPLICABLE and call.logical_op_key != branch.fork_op_key(call.tool):
            raise SpeculativeRollbackViolation(
                branch_id=branch.id,
                tool=call.tool,
                effect_class="deduplicable-but-branch-scoped-key",
                op_key=call.logical_op_key,
            )
        if effect_class == EFFECT_COMPENSABLE:
            ledger.schedule_compensation(call)
    branch.discard_internal_state()
    return True