# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-merge-a-speculative-agent-branch-that-has-already-mutated-external-state.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-merge-a-speculative-agent-branch-that-has-already-mutated-external-state.warrant.json
# Markdown SHA-256:   10e78a28034325a0e8b209acc99fa28a65c267d97174d2cd556b47619a493d7e
# Code AST SHA-256:   d4825d677ec692436cf787e3b242f51ef5f2c9c8a877e6d05e27d7bbfeb72bac
# Verified Classes:   UnsafeSpeculativeMergeError, Effect, Branch
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Iterable


class UnsafeSpeculativeMergeError(RuntimeError):
    """Raised when a merge would discard a branch with unrecorded external effects."""
    pass


@dataclass(frozen=True)
class Effect:
    effect_id: str
    target: str
    idempotency_key: str | None
    non_idempotent: bool


@dataclass
class Branch:
    name: str
    effects: list[Effect] = field(default_factory=list)


def merge_speculative_branches(
    survivor: Branch,
    discarded: Iterable[Branch],
) -> list[Effect]:
    survivor_keys = {
        e.idempotency_key for e in survivor.effects if e.idempotency_key
    }
    adopted: list[Effect] = []

    for branch in discarded:
        for effect in branch.effects:
            if effect.idempotency_key and effect.idempotency_key in survivor_keys:
                continue  # already represented; dedupe on key
            if effect.non_idempotent or effect.idempotency_key is None:
                raise UnsafeSpeculativeMergeError(
                    f"branch {branch.name!r} performed external write "
                    f"{effect.effect_id} to {effect.target!r} that is not in the "
                    f"survivor effect log; refusing to discard it silently. "
                    f"Run compensation or adopt the effect before merging."
                )
            adopted.append(effect)
            survivor_keys.add(effect.idempotency_key)

    survivor.effects.extend(adopted)
    return adopted


# Example: branch A created an unkeyed, non-idempotent ticket write.
survivor = Branch("B", effects=[
    Effect("e2", "ticketing", idempotency_key="ticket:T-1002", non_idempotent=False)
])
branch_a = Branch("A", effects=[
    Effect("e1", "ticketing", idempotency_key=None, non_idempotent=True)
])

merge_speculative_branches(survivor, [branch_a])  # raises UnsafeSpeculativeMergeError