# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-checkpoint-before-forking-a-speculative-branch-in-a-multi-agent-rollback-tree.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-checkpoint-before-forking-a-speculative-branch-in-a-multi-agent-rollback-tree.warrant.json
# Markdown SHA-256:   c798558df30220fcdb04b07c036df3af9a5b945eb5dbcf0d197112e0b96416b8
# Code AST SHA-256:   6f8d5a9ebc3c8540590d4490ccd60e545286ade6927597d69d0ed8842179d407
# Verified Classes:   ForkBarrierViolation, EffectLog
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class ForkBarrierViolation(RuntimeError):
    """Raised when a speculative fork is attempted before the effect-log barrier."""

    pass


class EffectLog:
    def __init__(self):
        self._entries = []
        self._barrier_seq = None

    def record(self, seq, effect):
        self._entries.append((seq, effect))

    def drain_barrier(self):
        # Barrier: all in-flight effects up to current seq are now durable.
        self._barrier_seq = len(self._entries)

    def effects_since(self, seq):
        return [e for s, e in self._entries if s >= seq]


def fork_branch(log: EffectLog, fork_seq: int, snapshot_seq: int) -> dict:
    """Fork a speculative branch only at a durable fork point.

    A durable fork requires that the effect-log barrier has advanced to at
    least the fork sequence, so that every effect attributable to the branch
    is contained in the snapshot's causal frontier.
    """
    if log._barrier_seq is None or log._barrier_seq < fork_seq:
        raise ForkBarrierViolation(
            f"fork_seq={fork_seq} precedes effect-log barrier "
            f"(barrier={log._barrier_seq}); branch effects would be orphaned "
            f"on rollback and invisible to convergent merge"
        )
    return {
        "fork_seq": fork_seq,
        "snapshot_seq": snapshot_seq,
        "effect_frontier": log.effects_since(snapshot_seq),
    }


def merge_aborted(branch: dict, log: EffectLog) -> list:
    """Enumerate an aborted branch's effects for subtraction during merge."""
    if branch.get("effect_frontier") is None:
        raise ForkBarrierViolation(
            "aborted branch has no recorded effect frontier; "
            "merge cannot attribute orphaned effects"
        )
    return [e for e in branch["effect_frontier"] if e in log.effects_since(branch["snapshot_seq"])]