# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-draw-the-line-between-compaction-induced-summary-loss-and-checkpoint-driven-recovery-in-a-lossy-auto-compaction-pipeline.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-draw-the-line-between-compaction-induced-summary-loss-and-checkpoint-driven-recovery-in-a-lossy-auto-compaction-pipeline.warrant.json
# Markdown SHA-256:   6f8a0cf312deddda002980e0f3e5e2c05cbe414e96a20b816270b43c2144cec7
# Code AST SHA-256:   3949d86fe8fc2cd07981df7df7210b4bc36c1df32b6cf9e3e05bb717c20504d4
# Verified Classes:   CompactionLineageError
# Defensive Guards:   1 fail-closed invariants (zero external deps)

class CompactionLineageError(RuntimeError):
    """Raised when a checkpoint cannot be safely used as a recovery target
    because it does not causally follow the last committed lossy compaction."""
    pass


def select_recovery_target(
    checkpoint_seq: int,
    last_committed_compaction_seq: int,
    compaction_in_flight: bool,
) -> str:
    """Return 'checkpoint' or 'replay' per the lineage boundary.

    A checkpoint is only a valid recovery target if it is strictly newer
    than the last committed lossy compaction AND no compaction is in flight.
    """
    if compaction_in_flight and checkpoint_seq <= last_committed_compaction_seq:
        # Torn state: checkpoint is concurrent with or older than an in-flight
        # lossy mutation. Restoring it would resurrect elided content.
        raise CompactionLineageError(
            f"checkpoint seq {checkpoint_seq} is not causally after "
            f"in-flight compaction seq {last_committed_compaction_seq}; "
            f"refusing restore, replay required"
        )

    if checkpoint_seq <= last_committed_compaction_seq:
        # Checkpoint predates the committed lossy compaction. It is a
        # pre-compaction artifact; replay from compaction output log instead.
        return "replay"

    return "checkpoint"


def recover(
    checkpoint_seq: int,
    last_committed_compaction_seq: int,
    compaction_in_flight: bool,
    restore_fn,
    replay_fn,
):
    target = select_recovery_target(
        checkpoint_seq, last_committed_compaction_seq, compaction_in_flight
    )
    if target == "checkpoint":
        return restore_fn()
    return replay_fn()