# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-hierarchical-compaction-rollback-to-a-lossy-checkpoint-when-post-checkpoint-deltas-are-non-idempotent.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-hierarchical-compaction-rollback-to-a-lossy-checkpoint-when-post-checkpoint-deltas-are-non-idempotent.warrant.json
# Markdown SHA-256:   f9bb35896bd34ead5a1e4199dc93f9f8dbe5ec0538c52449573a6d17bac63815
# Code AST SHA-256:   cb33d151e8a679abbc6fdc457f5a212a8017ba9311fcd52c581a0b2a8fc489b8
# Verified Classes:   UnsafeRollbackError, Checkpoint, EffectLedger
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class UnsafeRollbackError(RuntimeError):
    """Raised when rolling back to a lossy checkpoint would replay non-idempotent effects."""
    pass


class Checkpoint:
    def __init__(self, checkpoint_id, effect_watermark):
        self.checkpoint_id = checkpoint_id
        # Monotonic id of the last committed non-idempotent effect at snapshot time.
        # None means the watermark was compressed away and is no longer readable.
        self.effect_watermark = effect_watermark


class EffectLedger:
    def __init__(self):
        self.committed = []  # list of (effect_id, idempotent)

    def record(self, effect_id, idempotent):
        self.committed.append((effect_id, idempotent))

    def effects_after(self, watermark):
        if watermark is None:
            return list(self.committed)
        return [e for e in self.committed if e[0] > watermark]


def authorize_rollback(target: Checkpoint, ledger: EffectLedger, is_latest: bool) -> bool:
    if is_latest:
        return True

    pending = ledger.effects_after(target.effect_watermark)

    if not pending:
        return True

    # Watermark lost to compaction: we cannot prove which effects predate the checkpoint.
    if target.effect_watermark is None:
        raise UnsafeRollbackError(
            f"checkpoint {target.checkpoint_id}: effect watermark was compressed away; "
            f"{len(pending)} committed effect(s) cannot be ordered against the snapshot"
        )

    non_idempotent = [e for e in pending if not e[1]]
    if non_idempotent:
        raise UnsafeRollbackError(
            f"checkpoint {target.checkpoint_id}: rollback would replay "
            f"{len(non_idempotent)} non-idempotent effect(s): "
            f"{[e[0] for e in non_idempotent]}"
        )

    return True