# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-abort-a-speculative-branch-before-merge-when-side-effects-have-already-leaked-to-a-shared-external-sink.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-abort-a-speculative-branch-before-merge-when-side-effects-have-already-leaked-to-a-shared-external-sink.warrant.json
# Markdown SHA-256:   1b111a1069c6352b8e4d798b561dc554472a7abe271d5b582f474e54fd540f62
# Code AST SHA-256:   3b81ccf7a6f76830c0cddcb0a34e86c8c3a57aafc97c5b08467b2553df08e66e
# Verified Classes:   SpeculationLeakError, EffectLedger
# Defensive Guards:   5 fail-closed invariants (zero external deps)

class SpeculationLeakError(RuntimeError):
    """Raised when a speculative branch cannot be safely aborted because
    an emitted side effect violates the containment invariants C1-C4."""


class EffectLedger:
    PENDING = "pending"
    EMITTED = "emitted"
    COMPENSATED = "compensated"

    def __init__(self):
        self._entries = {}

    def register(self, effect_id, key, compensation):
        # C2: compensation must be registered before emission.
        if compensation is None:
            raise SpeculationLeakError(
                f"effect {effect_id} has no registered compensation; C2 violated"
            )
        self._entries[effect_id] = {
            "key": key,
            "state": self.PENDING,
            "compensation": compensation,
        }

    def mark_emitted(self, effect_id):
        self._entries[effect_id]["state"] = self.EMITTED

    def mark_compensated(self, effect_id):
        self._entries[effect_id]["state"] = self.COMPENSATED

    def is_abortable(self, branch_id, branch_effects):
        for eid in branch_effects:
            entry = self._entries.get(eid)
            if entry is None:
                raise SpeculationLeakError(
                    f"branch {branch_id} emitted unledgered effect {eid}"
                )
            if entry["state"] == self.EMITTED and entry["compensation"] is None:
                return False
        return True


def abort_branch(branch_id, branch_effects, ledger, sink):
    if not ledger.is_abortable(branch_id, branch_effects):
        # Do NOT abort. Quarantine and escalate.
        raise SpeculationLeakError(
            f"branch {branch_id} has uncompensable emitted effects; "
            f"graph must be marked dirty and reconciled, not aborted"
        )

    for eid in branch_effects:
        entry = ledger._entries[eid]
        if entry["state"] == ledger.EMITTED:
            try:
                sink.run_compensation(entry["compensation"])
            except Exception as exc:
                # C3: compensation failed; abort is unresolved.
                raise SpeculationLeakError(
                    f"compensation for {eid} failed: {exc}; "
                    f"branch {branch_id} is now an unresolved leak"
                ) from exc
            ledger.mark_compensated(eid)
    return True


def verify_merge_clean(winning_branch, all_branches, ledger):
    for bid, effects in all_branches.items():
        if bid == winning_branch:
            continue
        for eid in effects:
            state = ledger._entries[eid]["state"]
            if state == ledger.EMITTED:
                raise SpeculationLeakError(
                    f"non-winning branch {bid} left effect {eid} emitted; "
                    f"merge cannot be marked clean"
                )
    return True