# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-loss-of-speculative-side-effects-during-branch-reconciliation.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-loss-of-speculative-side-effects-during-branch-reconciliation.warrant.json
# Markdown SHA-256:   210ba6a9aad5b06ff532f6dd0e1934f068347a4bc8bfe177a79c06596cd4ed65
# Code AST SHA-256:   c689c0f36cd542cabf9d4e00f68a238862c347ba6b9654cb949c6d6eb310cac9
# Verified Classes:   UnresolvedSpeculativeEffect, BranchJournal, UnresolvedSpeculativeEffect, BranchJournal
# Defensive Guards:   8 fail-closed invariants (zero external deps)

class UnresolvedSpeculativeEffect(RuntimeError):
    pass


class BranchJournal:
    def __init__(self, branch_id):
        self.branch_id = branch_id
        self.entries = {}  # effect_id -> {"state": "intent"|"ack"|"aborted", "resource": str, "token": str}

    def write_intent(self, effect_id, resource, token):
        if effect_id in self.entries:
            raise UnresolvedSpeculativeEffect(
                f"effect {effect_id} already journaled on branch {self.branch_id}"
            )
        self.entries[effect_id] = {"state": "intent", "resource": resource, "token": token}

    def write_ack(self, effect_id):
        if effect_id not in self.entries:
            raise UnresolvedSpeculativeEffect(
                f"ack for unjournaled effect {effect_id} on branch {self.branch_id}"
            )
        self.entries[effect_id]["state"] = "ack"

    def unresolved_intents(self):
        return {eid: e for eid, e in self.entries.items() if e["state"] == "intent"}


def reconcile(branch_journals, committed_branch_id, compensator, effect_probe=None):
    # Invariant: no pruning until all intents are resolved.
    for j in branch_journals:
        if j.unresolved_intents():
            raise UnresolvedSpeculativeEffect(
                f"branch {j.branch_id} has unresolved intents; refusing to reconcile"
            )

    # Compensation set is derived from the UNION of all branch journals,
    # filtered by disposition -- never from the committed branch alone.
    compensation_set = []
    for j in branch_journals:
        if j.branch_id == committed_branch_id:
            continue
        for effect_id, entry in j.entries.items():
            if entry["state"] == "ack":
                compensation_set.append((effect_id, entry))
            elif entry["state"] == "intent":
                # Missing ack is NOT evidence of absence of effect.
                if effect_probe is not None and not effect_probe(entry["token"]):
                    continue
                compensation_set.append((effect_id, entry))

    for effect_id, entry in compensation_set:
        # Compensate by effect token, not resource key, so a committed
        # branch's reservation on the same resource is not released.
        compensator(entry["token"])


def commit_and_rollback(branch_journals, committed_branch_id, compensator):
    try:
        reconcile(branch_journals, committed_branch_id, compensator)
    except UnresolvedSpeculativeEffect:
        raise
    # Pruning is legal only after compensation completes.
    for j in branch_journals:
        if j.branch_id != committed_branch_id:
            j.entries.clear()

class UnresolvedSpeculativeEffect(RuntimeError):
    pass


class BranchJournal:
    def __init__(self, branch_id):
        self.branch_id = branch_id
        self.entries = {}

    def write_intent(self, effect_id, resource, token):
        if effect_id in self.entries:
            raise UnresolvedSpeculativeEffect(
                f"effect {effect_id} already journaled on branch {self.branch_id}"
            )
        self.entries[effect_id] = {"state": "intent", "resource": resource, "token": token}

    def write_ack(self, effect_id):
        if effect_id not in self.entries:
            raise UnresolvedSpeculativeEffect(
                f"ack for unjournaled effect {effect_id} on branch {self.branch_id}"
            )
        self.entries[effect_id]["state"] = "ack"

    def unresolved_intents(self):
        return {eid: e for eid, e in self.entries.items() if e["state"] == "intent"}


def reconcile(branch_journals, committed_branch_id, compensator, effect_probe=None):
    for j in branch_journals:
        if j.unresolved_intents():
            raise UnresolvedSpeculativeEffect(
                f"branch {j.branch_id} has unresolved intents; refusing to reconcile"
            )

    compensation_set = []
    for j in branch_journals:
        if j.branch_id == committed_branch_id:
            continue
        for effect_id, entry in j.entries.items():
            if entry["state"] == "ack":
                compensation_set.append((effect_id, entry))
            elif entry["state"] == "intent":
                if effect_probe is not None and not effect_probe(entry["token"]):
                    continue
                compensation_set.append((effect_id, entry))

    for effect_id, entry in compensation_set:
        compensator(entry["token"])


def commit_and_rollback(branch_journals, committed_branch_id, compensator):
    try:
        reconcile(branch_journals, committed_branch_id, compensator)
    except UnresolvedSpeculativeEffect:
        raise
    for j in branch_journals:
        if j.branch_id != committed_branch_id:
            j.entries.clear()