liminal-server 0.23.1

//! Pure checkpoint validation/recovery plans and saved tail starts.
use liminal::durability::{DurabilityError, DurableStore};

use super::fenced_attach_terminal::ComposedTerminalValidation;
use super::ops_session_replay::LogWalk;
use super::outbox_log::{OutboxLog, OutboxRestoreCursor};
use super::restore_checkpoint::{
    CheckpointError, CheckpointLiveState, CheckpointPositions, CheckpointSchema, RestoreCheckpoint,
};
use super::restore_checkpoint_locator::{
    CheckpointLocator, CheckpointLocatorError, checkpoint_conversation_route,
};

/// Reads authenticated index nodes through the same conversation route as the checkpoint.
pub(super) struct CheckpointHistoryReader<'a> {
    store: &'a dyn DurableStore,
}

impl<'a> CheckpointHistoryReader<'a> {
    pub(super) const fn new(store: &'a dyn DurableStore) -> Self {
        Self { store }
    }

    pub(super) async fn lookup(
        &self,
        root: &[u8; super::restore_checkpoint_history::ROOT_ENCODED_BYTES],
        key: super::restore_checkpoint_history_key::CheckpointHistoryKey,
    ) -> Result<
        super::restore_checkpoint_history_index::HistoryLookup,
        super::restore_checkpoint_history_index::HistoryIndexError,
    > {
        let root = super::restore_checkpoint_history::CheckpointHistoryRoot::decode_canonical(
            root,
            key.conversation_id(),
        )?;
        let route = checkpoint_conversation_route(key.conversation_id());
        super::restore_checkpoint_history_index::CheckpointHistoryIndex::new(&route, root)
            .lookup(self, key)
            .await
    }
}

#[async_trait::async_trait]
impl super::restore_checkpoint_history_index::HistoryNodeReader for CheckpointHistoryReader<'_> {
    async fn read_node(
        &self,
        conversation_route: &str,
        local_key: &[u8],
    ) -> Result<Option<Vec<u8>>, DurabilityError> {
        self.store.get_routed(conversation_route, local_key).await
    }
}

/// Validated checkpoint authority retained across restore and serving commits.
#[derive(Debug)]
pub(super) struct CheckpointMetadata {
    positions: CheckpointPositions,
    history_root: [u8; super::restore_checkpoint_history::ROOT_ENCODED_BYTES],
    prefix_digest: [u8; 32],
}

impl CheckpointMetadata {
    pub(super) fn validated(
        positions: CheckpointPositions,
        history_root: [u8; super::restore_checkpoint_history::ROOT_ENCODED_BYTES],
        prefix_digest: [u8; 32],
    ) -> Result<Self, CheckpointError> {
        positions.validate()?;
        super::restore_checkpoint_history::CheckpointHistoryRoot::decode_canonical(
            &history_root,
            positions.conversation_id,
        )
        .map_err(|source| CheckpointError::LiveState {
            source: Box::new(source),
        })?;
        Ok(Self {
            positions,
            history_root,
            prefix_digest,
        })
    }

    pub(super) fn with_history_root(
        self,
        root: [u8; super::restore_checkpoint_history::ROOT_ENCODED_BYTES],
    ) -> Result<Self, CheckpointError> {
        Self::validated(self.positions, root, self.prefix_digest)
    }

    /// Takes the locator witness paired with an already validated owner.
    pub(super) fn from_checkpoint<L: CheckpointLiveState>(
        checkpoint: &RestoreCheckpoint<
            L,
            { super::restore_checkpoint_history::ROOT_ENCODED_BYTES },
        >,
        locator: &CheckpointLocator,
    ) -> Result<Self, CheckpointError> {
        let positions = checkpoint.positions();
        locator.validate_positions(&positions)?;
        Self::validated(
            positions,
            *checkpoint.history_root(),
            *locator.prefix_digest(),
        )
    }

    /// Takes the witness computed over the complete authoritative rebuild.
    pub(super) fn from_rebuilt<L: CheckpointLiveState>(
        checkpoint: &RestoreCheckpoint<
            L,
            { super::restore_checkpoint_history::ROOT_ENCODED_BYTES },
        >,
        prefix_digest: [u8; 32],
    ) -> Result<Self, CheckpointError> {
        Self::validated(
            checkpoint.positions(),
            *checkpoint.history_root(),
            prefix_digest,
        )
    }

    pub(super) const fn schema(&self) -> CheckpointSchema {
        self.positions.schema
    }
    pub(super) const fn history_root(
        &self,
    ) -> &[u8; super::restore_checkpoint_history::ROOT_ENCODED_BYTES] {
        &self.history_root
    }
    pub(super) const fn prefix_digest(&self) -> [u8; 32] {
        self.prefix_digest
    }
    pub(super) const fn merged_base_sequence(&self) -> u64 {
        self.positions.merged_base_sequence
    }
    pub(super) const fn merged_extension_sequence(&self) -> u64 {
        self.positions.merged_extension_sequence
    }

    fn advanced(
        &self,
        positions: CheckpointPositions,
        history_root: [u8; super::restore_checkpoint_history::ROOT_ENCODED_BYTES],
        prefix_digest: [u8; 32],
    ) -> Result<Self, CheckpointError> {
        if positions.conversation_id != self.positions.conversation_id {
            return Err(CheckpointError::ForeignConversation {
                expected: self.positions.conversation_id,
                actual: positions.conversation_id,
            });
        }
        if positions.next_log_sequence < self.positions.next_log_sequence
            || positions.extension_sequence < self.positions.extension_sequence
            || positions.merged_base_sequence < self.positions.merged_base_sequence
            || positions.merged_extension_sequence < self.positions.merged_extension_sequence
        {
            return Err(CheckpointError::Position {
                field: "checkpoint metadata regression",
            });
        }
        match (self.positions.schema, positions.schema) {
            (
                CheckpointSchema::V4Suffix,
                CheckpointSchema::V2Prefix | CheckpointSchema::V3Suffix,
            )
            | (CheckpointSchema::V3Suffix, CheckpointSchema::V2Prefix) => {
                return Err(CheckpointError::Association {
                    field: "checkpoint schema regression",
                });
            }
            _ => {}
        }
        Self::validated(positions, history_root, prefix_digest)
    }
}

impl super::state::ConversationAuthority {
    pub(super) fn checkpoint_metadata(
        &self,
    ) -> Result<&CheckpointMetadata, super::state::StateError> {
        self.checkpoint_metadata.as_ref().ok_or_else(|| {
            super::state::StateError::invariant("validated checkpoint metadata is absent")
        })
    }

    fn validate_checkpoint_metadata(
        &self,
        metadata: &CheckpointMetadata,
    ) -> Result<(), CheckpointError> {
        let actual = self.checkpoint_positions(
            metadata.schema(),
            metadata.merged_base_sequence(),
            metadata.merged_extension_sequence(),
        )?;
        if actual != metadata.positions {
            return Err(CheckpointError::Association {
                field: "live checkpoint metadata",
            });
        }
        Ok(())
    }

    /// Installs only metadata that matches the current move-owned authority.
    pub(super) fn install_checkpoint_metadata(
        &mut self,
        metadata: CheckpointMetadata,
    ) -> Result<(), CheckpointError> {
        self.validate_checkpoint_metadata(&metadata)?;
        self.checkpoint_metadata = Some(metadata);
        Ok(())
    }

    /// Records the validated result only after its durable batch has committed.
    pub(super) fn advance_checkpoint_metadata(
        &mut self,
        positions: CheckpointPositions,
        history_root: [u8; super::restore_checkpoint_history::ROOT_ENCODED_BYTES],
        prefix_digest: [u8; 32],
    ) -> Result<(), CheckpointError> {
        let prior = self
            .checkpoint_metadata
            .as_ref()
            .ok_or(CheckpointError::Association {
                field: "absent checkpoint metadata",
            })?;
        let next = prior.advanced(positions, history_root, prefix_digest)?;
        self.install_checkpoint_metadata(next)
    }
}

/// Move-owned restore inputs derived from one jointly validated checkpoint.
pub(super) struct PreparedCheckpointRestore {
    pub(super) owner: super::state::ConversationAuthority,
    pub(super) start: TailStart,
    pub(super) audit: ComposedTerminalValidation,
    pub(super) root: super::restore_checkpoint_history::CheckpointHistoryRoot,
    pub(super) prefix_digest: [u8; 32],
}

/// Retains the actual framing root and saved audit before tail processing.
pub(super) fn prepare_checkpoint_restore(
    checkpoint: RestoreCheckpoint<
        super::state::ConversationAuthority,
        { super::restore_checkpoint_history::ROOT_ENCODED_BYTES },
    >,
    locator: &CheckpointLocator,
    config: &crate::config::types::ParticipantConfig,
) -> Result<PreparedCheckpointRestore, CheckpointError> {
    let metadata = CheckpointMetadata::from_checkpoint(&checkpoint, locator)?;
    let start = TailStart::from_checkpoint(&checkpoint);
    let (positions, root_bytes, mut owner) = checkpoint.into_parts();
    let root = super::restore_checkpoint_history::CheckpointHistoryRoot::decode_canonical(
        &root_bytes,
        positions.conversation_id,
    )
    .map_err(|source| CheckpointError::LiveState {
        source: Box::new(source),
    })?;
    let audit = match owner.checkpoint_audit_source.take() {
        Some(bytes) => ComposedTerminalValidation::decode_checkpoint(
            &bytes,
            config.identity_slots,
            positions.covered_log_sequence,
        )?,
        None if positions.covered_log_sequence == 0
            && owner.obligation_debt_dispatch.is_none()
            && root
                == super::restore_checkpoint_history::CheckpointHistoryRoot::empty(
                    positions.conversation_id,
                ) =>
        {
            ComposedTerminalValidation::for_identity_slots(config.identity_slots).map_err(
                |source| CheckpointError::LiveState {
                    source: Box::new(source),
                },
            )?
        }
        None => {
            return Err(CheckpointError::Association {
                field: "absent saved terminal audit",
            });
        }
    };
    owner.retain_checkpoint_metadata(root, &audit)?;
    owner.install_checkpoint_metadata(metadata)?;
    Ok(PreparedCheckpointRestore {
        owner,
        start,
        audit,
        root,
        prefix_digest: *locator.prefix_digest(),
    })
}

/// Captures the validated merge result using the caller's actual HISTORY root.
pub(super) fn prepare_replayed_checkpoint(
    mut owner: super::state::ConversationAuthority,
    audit: &ComposedTerminalValidation,
    schema: CheckpointSchema,
    root: super::restore_checkpoint_history::CheckpointHistoryRoot,
    prefix_digest: [u8; 32],
) -> Result<
    RestoreCheckpoint<
        super::state::ConversationAuthority,
        { super::restore_checkpoint_history::ROOT_ENCODED_BYTES },
    >,
    CheckpointError,
> {
    let extension_sequence = owner
        .outbox
        .as_ref()
        .ok_or(CheckpointError::Position {
            field: "absent replayed outbox",
        })?
        .next_extension_sequence();
    let positions =
        owner.checkpoint_positions(schema, owner.next_log_sequence, extension_sequence)?;
    owner.retain_checkpoint_metadata(root, audit)?;
    let metadata =
        CheckpointMetadata::validated(positions, root.encode_canonical(), prefix_digest)?;
    owner.install_checkpoint_metadata(metadata)?;
    RestoreCheckpoint::new(positions, root.encode_canonical(), owner)
}

/// Tail replay keeps the history provider's original typed failure.
#[derive(Debug, thiserror::Error)]
pub(super) enum TailReplayError<E> {
    #[error(transparent)]
    Replay(#[from] super::outbox_replay::RestoreError),
    #[error("checkpoint source read failed: {0}")]
    Source(#[source] E),
}
impl<E: std::error::Error + 'static> From<super::state::StateError> for TailReplayError<E> {
    fn from(error: super::state::StateError) -> Self {
        Self::Replay(error.into())
    }
}

pub(super) enum CheckpointDecision<L, const N: usize> {
    Restore(RestoreCheckpoint<L, N>),
    Rebuild(RebuildRequired),
}
#[derive(Debug, thiserror::Error)]
pub(super) enum CheckpointRefusal {
    #[error("checkpoint locator is absent")]
    Missing,
    #[error("checkpoint body is absent")]
    MissingBody,
    #[error("checkpoint locator refused: {0}")]
    Locator(#[source] CheckpointLocatorError),
    #[error("checkpoint refused: {0}")]
    Invalid(#[source] CheckpointError),
}
pub(super) struct RebuildRequired {
    pub(super) conversation_id: u64,
    pub(super) refusal: CheckpointRefusal,
}
pub(super) struct FreshCheckpoint<L, const N: usize> {
    pub(super) refusal: CheckpointRefusal,
    pub(super) checkpoint: RestoreCheckpoint<L, N>,
}

impl RebuildRequired {
    /// Validates rebuilt authority and preserves the diagnostic for publication.
    /// The caller publishes through the atomic batch after reporting the refusal.
    pub(super) fn prepare_fresh<L: CheckpointLiveState, const N: usize>(
        self,
        positions: CheckpointPositions,
        root: [u8; N],
        live: L,
    ) -> Result<FreshCheckpoint<L, N>, CheckpointError> {
        if positions.conversation_id != self.conversation_id {
            return Err(CheckpointError::ForeignConversation {
                expected: self.conversation_id,
                actual: positions.conversation_id,
            });
        }
        let checkpoint = RestoreCheckpoint::new(positions, root, live)?;
        Ok(FreshCheckpoint {
            refusal: self.refusal,
            checkpoint,
        })
    }
}
/// Validates one checkpoint and delegates locator association to its owning codec.
/// The refusal is an explicit recovery plan, never a silent history fallback.
pub(super) fn validate_or_plan_rebuild<L: CheckpointLiveState, const N: usize, D, C>(
    decoded_locator: &D,
    conversation_id: u64,
    bytes: Option<&[u8]>,
    make_context: impl FnOnce(&CheckpointPositions, &[u8; N]) -> Result<C, CheckpointError>,
    validate_locator: impl FnOnce(&D, &CheckpointPositions, &[u8; N]) -> Result<(), CheckpointError>,
) -> CheckpointDecision<L, N>
where
    C: std::borrow::Borrow<L::RestoreContext>,
{
    let Some(bytes) = bytes else {
        return CheckpointDecision::Rebuild(RebuildRequired {
            conversation_id,
            refusal: CheckpointRefusal::Missing,
        });
    };
    let validated = (|| {
        let (positions, root, _) = super::restore_checkpoint_codec::decode(bytes, conversation_id)?;
        validate_locator(decoded_locator, &positions, &root)?;
        let context = make_context(&positions, &root)?;
        RestoreCheckpoint::decode_canonical(bytes, conversation_id, context.borrow())
    })();
    match validated {
        Ok(checkpoint) => CheckpointDecision::Restore(checkpoint),
        Err(error) => CheckpointDecision::Rebuild(RebuildRequired {
            conversation_id,
            refusal: CheckpointRefusal::Invalid(error),
        }),
    }
}
/// Verifies the routed body's content identity before the live owner is decoded.
/// Root association remains the responsibility of the qualified history codec.
pub(super) fn validate_locator_and_body<L: CheckpointLiveState, const N: usize, C>(
    locator: &CheckpointLocator,
    conversation_id: u64,
    body: Option<&[u8]>,
    make_context: impl FnOnce(&CheckpointPositions, &[u8; N]) -> Result<C, CheckpointError>,
    validate_root: impl FnOnce(&CheckpointLocator, &[u8; N]) -> Result<(), CheckpointError>,
) -> CheckpointDecision<L, N>
where
    C: std::borrow::Borrow<L::RestoreContext>,
{
    let refusal = |refusal| {
        CheckpointDecision::Rebuild(RebuildRequired {
            conversation_id,
            refusal,
        })
    };
    if locator.conversation_id() != conversation_id {
        return refusal(CheckpointRefusal::Invalid(
            CheckpointError::ForeignConversation {
                expected: conversation_id,
                actual: locator.conversation_id(),
            },
        ));
    }
    let Some(body) = body else {
        return refusal(CheckpointRefusal::MissingBody);
    };
    if let Err(error) = locator.body_key().verify_body(body) {
        return refusal(CheckpointRefusal::Locator(error));
    }
    let validated = (|| {
        let (positions, root, _) = super::restore_checkpoint_codec::decode(body, conversation_id)?;
        locator.validate_positions(&positions)?;
        validate_root(locator, &root)?;
        let context = make_context(&positions, &root)?;
        locator.decode_checkpoint_body(body, context.borrow(), |_, _, _| Ok(()))
    })();
    match validated {
        Ok(checkpoint) => CheckpointDecision::Restore(checkpoint),
        Err(error) => refusal(CheckpointRefusal::Invalid(error)),
    }
}

/// One routed lookup result, with the original locator retained for recovery CAS.
pub(super) struct LoadedCheckpoint<L, const N: usize> {
    pub(super) decision: CheckpointDecision<L, N>,
    pub(super) observed_locator: Option<Vec<u8>>,
}

/// Loads only the locator and its content-addressed body.
///
/// Store failures propagate unchanged. Missing or invalid checkpoint bytes
/// produce an explicit rebuild decision; they never install a live owner.
pub(super) async fn load_checkpoint<L: CheckpointLiveState, const N: usize>(
    store: &dyn DurableStore,
    conversation_id: u64,
    make_context: impl FnOnce(
        &CheckpointPositions,
        &[u8; N],
    ) -> Result<L::RestoreContext, CheckpointError>
    + Send,
    validate_root: impl FnOnce(&CheckpointLocator, &[u8; N]) -> Result<(), CheckpointError> + Send,
) -> Result<LoadedCheckpoint<L, N>, DurabilityError> {
    let route = checkpoint_conversation_route(conversation_id);
    let observed_locator = store
        .get_routed(&route, CheckpointLocator::routed_key())
        .await?;
    let refusal = |refusal, observed_locator| LoadedCheckpoint {
        decision: CheckpointDecision::Rebuild(RebuildRequired {
            conversation_id,
            refusal,
        }),
        observed_locator,
    };
    let Some(bytes) = observed_locator.as_deref() else {
        return Ok(refusal(CheckpointRefusal::Missing, observed_locator));
    };
    let locator = match CheckpointLocator::decode_canonical(bytes, conversation_id) {
        Ok(locator) => locator,
        Err(error) => {
            return Ok(refusal(CheckpointRefusal::Locator(error), observed_locator));
        }
    };
    let body = super::checkpoint_body_parts::read_body(
        store,
        &locator.conversation_route(),
        locator.body_key(),
    )
    .await?;
    let decision = validate_locator_and_body(
        &locator,
        conversation_id,
        body.as_deref(),
        make_context,
        validate_root,
    );
    Ok(LoadedCheckpoint {
        decision,
        observed_locator,
    })
}

/// Tail starts obtained only from a validated checkpoint.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct TailStart {
    operation_sequence: u64,
    extension_sequence: u64,
    schema: CheckpointSchema,
}
impl TailStart {
    pub(super) const fn from_checkpoint<L: CheckpointLiveState, const N: usize>(
        checkpoint: &RestoreCheckpoint<L, N>,
    ) -> Self {
        let positions = checkpoint.positions();
        Self {
            operation_sequence: positions.next_log_sequence,
            extension_sequence: positions.extension_sequence,
            schema: positions.schema,
        }
    }
    pub(super) const fn operation_sequence(self) -> u64 {
        self.operation_sequence
    }
    pub(super) const fn extension_sequence(self) -> u64 {
        self.extension_sequence
    }
    pub(super) const fn schema(self) -> CheckpointSchema {
        self.schema
    }
    pub(super) const fn operation_walk(self, audit: ComposedTerminalValidation) -> LogWalk {
        LogWalk::from_checkpoint_start(self, audit)
    }
    pub(super) fn extension_cursor<'a>(
        self,
        log: &'a OutboxLog,
        route: &str,
    ) -> OutboxRestoreCursor<'a> {
        log.restore_routed_cursor_from(route, self.extension_sequence)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::error::Error;
    #[derive(Debug)]
    struct Live(u8);
    impl CheckpointLiveState for Live {
        type RestoreContext = u8;
        fn encode_live(&self) -> Result<Vec<u8>, CheckpointError> {
            Ok(vec![self.0])
        }
        fn decode_live(
            bytes: &[u8],
            positions: &CheckpointPositions,
            history_root: &[u8],
            context: &Self::RestoreContext,
        ) -> Result<Self, CheckpointError> {
            if *context != 7 {
                return Err(CheckpointError::LiveState {
                    source: std::io::Error::other("invalid restore configuration").into(),
                });
            }
            let value = bytes
                .first()
                .copied()
                .ok_or(CheckpointError::LengthOverflow)?;
            let live = Self(value);
            live.validate_live(positions, history_root)?;
            Ok(live)
        }
        fn validate_live(
            &self,
            positions: &CheckpointPositions,
            history_root: &[u8],
        ) -> Result<(), CheckpointError> {
            if self.0 != 7 || positions.conversation_id != 7 || history_root != [9; 32] {
                return Err(CheckpointError::LiveState {
                    source: std::io::Error::other("invalid live fixture identity").into(),
                });
            }
            Ok(())
        }
    }
    fn positions() -> CheckpointPositions {
        CheckpointPositions {
            conversation_id: 7,
            covered_log_sequence: 999,
            next_event_ordinal: 1,
            next_log_sequence: 1000,
            next_seq: 1,
            next_order: 1,
            next_participant: 1,
            extension_sequence: 2000,
            highest_delivery_seq: 0,
            observer_progress: 0,
            merged_base_sequence: 1000,
            merged_extension_sequence: 2000,
            schema: CheckpointSchema::V4Suffix,
        }
    }
    fn bytes() -> Result<Vec<u8>, CheckpointError> {
        RestoreCheckpoint::new(positions(), [9; 32], Live(7))?.encode_canonical()
    }
    fn check_locator(
        locator: u64,
        saved: &CheckpointPositions,
        root: &[u8; 32],
    ) -> Result<(), CheckpointError> {
        if locator != saved.next_log_sequence || root != &[9; 32] {
            return Err(CheckpointError::Position {
                field: "locator checkpoint boundary",
            });
        }
        Ok(())
    }
    #[test]
    fn valid_checkpoint_carries_both_saved_tail_heads() -> Result<(), Box<dyn Error>> {
        let encoded = bytes()?;
        let CheckpointDecision::Restore(checkpoint) = validate_or_plan_rebuild::<Live, 32, _, _>(
            &1000,
            7,
            Some(&encoded),
            |_, _| Ok(&7),
            |locator, saved, root| check_locator(*locator, saved, root),
        ) else {
            return Err("valid checkpoint refused".into());
        };
        let start = TailStart::from_checkpoint(&checkpoint);
        assert_eq!(start.operation_sequence, 1000);
        assert_eq!(start.extension_sequence, 2000);
        assert_eq!(start.schema, CheckpointSchema::V4Suffix);
        Ok(())
    }
    #[test]
    fn corrupted_checkpoint_requires_loud_rebuild_and_fresh_publication()
    -> Result<(), Box<dyn Error>> {
        let mut encoded = bytes()?;
        *encoded.last_mut().ok_or("empty checkpoint")? ^= 1;
        let CheckpointDecision::Rebuild(required) = validate_or_plan_rebuild::<Live, 32, _, _>(
            &1000,
            7,
            Some(&encoded),
            |_, _| Ok(&7),
            |locator, saved, root| check_locator(*locator, saved, root),
        ) else {
            return Err("corrupt checkpoint accepted".into());
        };
        assert!(matches!(
            required.refusal,
            CheckpointRefusal::Invalid(CheckpointError::Integrity)
        ));
        let rebuilt = required.prepare_fresh(positions(), [9; 32], Live(7))?;
        assert!(matches!(
            rebuilt.refusal,
            CheckpointRefusal::Invalid(CheckpointError::Integrity)
        ));
        let canonical = rebuilt.checkpoint.encode_canonical()?;
        assert_ne!(canonical, encoded);
        assert!(RestoreCheckpoint::<Live, 32>::decode_canonical(&canonical, 7, &7).is_ok());
        Ok(())
    }
    #[test]
    fn missing_checkpoint_and_bad_locator_cannot_become_ready() -> Result<(), Box<dyn Error>> {
        assert!(matches!(
            validate_or_plan_rebuild::<Live, 32, _, _>(
                &1000,
                7,
                None,
                |_, _| Ok(&7),
                |locator, saved, root| check_locator(*locator, saved, root)
            ),
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Missing,
                ..
            })
        ));
        let encoded = bytes()?;
        assert!(matches!(
            validate_or_plan_rebuild::<Live, 32, _, _>(
                &1001,
                7,
                Some(&encoded),
                |_, _| Ok(&7),
                |locator, saved, root| check_locator(*locator, saved, root)
            ),
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Invalid(CheckpointError::Position { .. }),
                ..
            })
        ));
        Ok(())
    }
    #[test]
    fn rebuilt_owner_must_validate_before_fresh_publication() -> Result<(), Box<dyn Error>> {
        let CheckpointDecision::Rebuild(required) = validate_or_plan_rebuild::<Live, 32, _, _>(
            &1000,
            7,
            None,
            |_, _| Ok(&7),
            |locator, saved, root| check_locator(*locator, saved, root),
        ) else {
            return Err("missing checkpoint accepted".into());
        };
        assert!(matches!(
            required.prepare_fresh(positions(), [9; 32], Live(8)),
            Err(CheckpointError::LiveState { .. })
        ));
        Ok(())
    }
    #[test]
    fn invalid_restore_context_is_an_explicit_owner_refusal() -> Result<(), Box<dyn Error>> {
        let encoded = bytes()?;
        assert!(matches!(
            validate_or_plan_rebuild::<Live, 32, _, _>(
                &1000,
                7,
                Some(&encoded),
                |_, _| Ok(&8),
                |locator, saved, root| check_locator(*locator, saved, root)
            ),
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Invalid(CheckpointError::LiveState { .. }),
                ..
            })
        ));
        Ok(())
    }

    #[test]
    fn tail_starts_keep_every_validated_schema_phase() -> Result<(), Box<dyn Error>> {
        for schema in [
            CheckpointSchema::V2Prefix,
            CheckpointSchema::V3Suffix,
            CheckpointSchema::V4Suffix,
        ] {
            let mut saved = positions();
            saved.schema = schema;
            let checkpoint = RestoreCheckpoint::new(saved, [9; 32], Live(7))?;
            let start = TailStart::from_checkpoint(&checkpoint);
            assert_eq!(start.operation_sequence(), saved.covered_log_sequence + 1);
            assert_eq!(start.extension_sequence(), saved.extension_sequence);
            assert_eq!(start.schema(), schema);
        }
        Ok(())
    }

    struct HistoryLookupLive(Live);

    impl CheckpointLiveState for HistoryLookupLive {
        type RestoreContext = std::cell::Cell<u64>;

        fn encode_live(&self) -> Result<Vec<u8>, CheckpointError> {
            self.0.encode_live()
        }

        fn decode_live(
            bytes: &[u8],
            positions: &CheckpointPositions,
            history_root: &[u8],
            context: &Self::RestoreContext,
        ) -> Result<Self, CheckpointError> {
            context.set(1);
            Live::decode_live(bytes, positions, history_root, &7).map(Self)
        }

        fn validate_live(
            &self,
            positions: &CheckpointPositions,
            history_root: &[u8],
        ) -> Result<(), CheckpointError> {
            self.0.validate_live(positions, history_root)
        }
    }

    #[test]
    fn locator_refusal_precedes_live_history_lookup() -> Result<(), Box<dyn Error>> {
        let input = bytes()?;
        let history_lookups = std::cell::Cell::new(0);
        let decision = validate_or_plan_rebuild::<HistoryLookupLive, 32, _, _>(
            &1000,
            7,
            Some(&input),
            |_, _| Ok(&history_lookups),
            |_, _, _| {
                Err(CheckpointError::Position {
                    field: "checkpoint locator",
                })
            },
        );
        assert!(matches!(
            decision,
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Invalid(CheckpointError::Position {
                    field: "checkpoint locator"
                }),
                ..
            })
        ));
        assert_eq!(
            history_lookups.get(),
            0,
            "locator refusal must precede history authority access"
        );
        Ok(())
    }

    fn concrete_locator(body: &[u8]) -> Result<CheckpointLocator, CheckpointLocatorError> {
        use super::super::restore_checkpoint_locator::CheckpointBodyKey;
        CheckpointLocator::new(7, 999, 2000, [3; 32], CheckpointBodyKey::for_body(body))
    }

    #[test]
    fn missing_or_changed_body_never_decodes_live_authority() -> Result<(), Box<dyn Error>> {
        let body = bytes()?;
        let locator = concrete_locator(&body)?;
        let history_lookups = std::cell::Cell::new(0);
        let missing = validate_locator_and_body::<HistoryLookupLive, 32, _>(
            &locator,
            7,
            None,
            |_, _| Ok(&history_lookups),
            |_, _| Ok(()),
        );
        assert!(matches!(
            missing,
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::MissingBody,
                ..
            })
        ));
        let changed = validate_locator_and_body::<HistoryLookupLive, 32, _>(
            &locator,
            7,
            Some(b"changed"),
            |_, _| Ok(&history_lookups),
            |_, _| Ok(()),
        );
        assert!(matches!(
            changed,
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Locator(CheckpointLocatorError::BodyDigest),
                ..
            })
        ));
        assert_eq!(history_lookups.get(), 0);
        Ok(())
    }

    #[test]
    fn locator_body_preserves_the_qualified_root_schema() -> Result<(), Box<dyn Error>> {
        let body = bytes()?;
        let locator = concrete_locator(&body)?;
        let history_lookups = std::cell::Cell::new(0);
        let root_validations = std::cell::Cell::new(0);
        let decision = validate_locator_and_body::<HistoryLookupLive, 32, _>(
            &locator,
            7,
            Some(&body),
            |_, _| Ok(&history_lookups),
            |actual, root| {
                assert_eq!(actual, &locator);
                assert_eq!(root, &[9; 32]);
                root_validations.set(root_validations.get() + 1);
                Ok(())
            },
        );
        let CheckpointDecision::Restore(checkpoint) = decision else {
            return Err("valid locator body was refused".into());
        };
        assert_eq!(root_validations.get(), 1);
        assert_eq!(history_lookups.get(), 1);
        assert_eq!(checkpoint.positions(), positions());
        assert_eq!(checkpoint.live().0.0, 7);
        Ok(())
    }

    #[test]
    fn routed_loader_preserves_absent_and_malformed_locator_rebuild_authority()
    -> Result<(), Box<dyn Error>> {
        use super::super::tests_memory_store::MemoryStore;
        use liminal::durability::{RoutedBatchItem, bridge::block_on};

        let store = MemoryStore::default();
        let missing = block_on(load_checkpoint::<Live, 32>(
            &store,
            7,
            |_, _| Ok(7),
            |_, _| Ok(()),
        ))??;
        assert!(matches!(
            missing.decision,
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Missing,
                ..
            })
        ));
        assert_eq!(missing.observed_locator, None);
        let invalid = b"malformed locator".to_vec();
        block_on(store.apply_routed_batch(
            &checkpoint_conversation_route(7),
            vec![RoutedBatchItem {
                key: CheckpointLocator::routed_key().to_vec(),
                expected: None,
                value: invalid.clone(),
                ttl: None,
            }],
        ))??;
        let refused = block_on(load_checkpoint::<Live, 32>(
            &store,
            7,
            |_, _| Err(CheckpointError::LiveEncoding),
            |_, _| Err(CheckpointError::LiveEncoding),
        ))??;
        assert!(matches!(
            refused.decision,
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Locator(_),
                ..
            })
        ));
        assert_eq!(refused.observed_locator, Some(invalid));
        Ok(())
    }

    #[test]
    fn routed_loader_reads_only_the_locator_and_exact_body() -> Result<(), Box<dyn Error>> {
        use super::super::tests_memory_store::MemoryStore;
        use liminal::durability::{CountingStore, RoutedBatchItem, bridge::block_on};

        let store = CountingStore::new(MemoryStore::default());
        let body = bytes()?;
        let locator = concrete_locator(&body)?;
        let encoded_locator = locator.encode_canonical().to_vec();
        block_on(store.apply_routed_batch(
            &checkpoint_conversation_route(7),
            vec![
                RoutedBatchItem {
                    key: CheckpointLocator::routed_key().to_vec(),
                    expected: None,
                    value: encoded_locator.clone(),
                    ttl: None,
                },
                RoutedBatchItem {
                    key: locator.body_key().routed_key(),
                    expected: None,
                    value: body,
                    ttl: None,
                },
            ],
        ))??;
        let before = store.snapshot();
        let associated = std::sync::atomic::AtomicBool::new(false);
        let loaded = block_on(load_checkpoint::<Live, 32>(
            &store,
            7,
            |saved, root| {
                assert!(associated.load(std::sync::atomic::Ordering::Relaxed));
                assert_eq!(*saved, positions());
                assert_eq!(root, &[9; 32]);
                Ok(7)
            },
            |actual, root| {
                assert_eq!(actual, &locator);
                assert_eq!(root, &[9; 32]);
                associated.store(true, std::sync::atomic::Ordering::Relaxed);
                Ok(())
            },
        ))??;
        let CheckpointDecision::Restore(checkpoint) = loaded.decision else {
            return Err("canonical routed checkpoint refused".into());
        };
        assert_eq!(loaded.observed_locator, Some(encoded_locator));
        assert_eq!(checkpoint.positions(), positions());
        assert_eq!(checkpoint.live().0, 7);
        let reads = store.snapshot().since(&before);
        assert_eq!(reads.get_routed_calls, 2);
        assert_eq!(reads.get_routed_rows, 2);
        assert_eq!(reads.read_from_calls, 0);
        assert_eq!(reads.read_at_calls, 0);
        assert_eq!(reads.scan_page_calls, 0);
        assert_eq!(reads.flushes, 0);
        assert_eq!(reads.apply_routed_batch_calls, 0);
        let constructed = std::sync::atomic::AtomicBool::new(false);
        let refused = block_on(load_checkpoint::<Live, 32>(
            &store,
            7,
            |_, _| {
                constructed.store(true, std::sync::atomic::Ordering::Relaxed);
                Ok(7)
            },
            |_, _| {
                Err(CheckpointError::Association {
                    field: "history root",
                })
            },
        ))??;
        assert!(!constructed.load(std::sync::atomic::Ordering::Relaxed));
        assert!(matches!(
            refused.decision,
            CheckpointDecision::Rebuild(RebuildRequired {
                refusal: CheckpointRefusal::Invalid(CheckpointError::Association {
                    field: "history root"
                }),
                ..
            })
        ));
        assert_eq!(refused.observed_locator, loaded.observed_locator);
        Ok(())
    }

    #[test]
    fn checkpoint_envelope_component_reads_equal_rows_and_bytes() -> Result<(), Box<dyn Error>> {
        use crate::boot_history_harness::{
            ReadCounts, ReadSource, StageReads, during_stage, history_pair,
        };
        use liminal::durability::bridge::block_on;
        use std::collections::BTreeMap;

        const HISTORY: &str = "fixture:checkpoint-envelope-history";
        const BODIES: &str = "fixture:checkpoint-envelope-bodies";
        const HEAD: &str = "fixture:checkpoint-envelope-head";
        const STAGE: &str = "checkpoint-envelope-component";
        let payload = bytes()?;
        let body_bytes = u64::try_from(payload.len())?;
        let pair = history_pair("checkpoint-envelope-context-v1", |store, range| {
            let next = range.end;
            let first = range.start;
            let rows = range
                .map(|sequence| sequence.to_le_bytes().to_vec())
                .collect();
            assert_eq!(block_on(store.append_all(HISTORY, rows, first))??, next);
            let head = block_on(store.read_value(HEAD))??.unwrap_or_default();
            assert_eq!(
                block_on(store.append_all(BODIES, vec![payload.clone()], head))??,
                head + 1
            );
            block_on(store.cas(HEAD, head, head + 1))??;
            Ok(())
        })?;
        let expected: StageReads = BTreeMap::from([(
            STAGE.to_owned(),
            BTreeMap::from([
                (
                    ReadSource::Value(HEAD.to_owned()),
                    ReadCounts { rows: 1, bytes: 8 },
                ),
                (
                    ReadSource::Stream(BODIES.to_owned()),
                    ReadCounts {
                        rows: 1,
                        bytes: body_bytes,
                    },
                ),
            ]),
        )]);
        let measured = pair.assert_stage(STAGE, &expected, |store| {
            during_stage(STAGE, || {
                let head =
                    block_on(store.read_value(HEAD))??.ok_or("checkpoint fixture head absent")?;
                let body = block_on(store.read_at(
                    BODIES,
                    head.checked_sub(1).ok_or("checkpoint fixture head zero")?,
                ))??
                .ok_or("checkpoint fixture body absent")?;
                let CheckpointDecision::Restore(checkpoint) =
                    validate_or_plan_rebuild::<Live, 32, _, _>(
                        &1000,
                        7,
                        Some(&body.payload),
                        |_, _| Ok(&7),
                        |locator, saved, root| check_locator(*locator, saved, root),
                    )
                else {
                    return Err("checkpoint fixture envelope refused".into());
                };
                assert_eq!(checkpoint.live().0, 7);
                let start = TailStart::from_checkpoint(&checkpoint);
                assert_eq!(start.operation_sequence(), 1000);
                assert_eq!(start.extension_sequence(), 2000);
                Ok(())
            })
        })?;
        println!("checkpoint envelope measured reads: {measured:?}");
        Ok(())
    }

    #[test]
    fn extension_tail_component_is_pinned_by_shared_history_harness() -> Result<(), Box<dyn Error>>
    {
        use super::super::outbox_log::{OUTBOX_STREAM_PREFIX, OutboxRow, encode_row};
        use crate::boot_history_harness::{
            ReadCounts, ReadSource, StageReads, during_stage, history_pair,
        };
        use liminal::durability::bridge::block_on;
        use std::collections::BTreeMap;

        const HEAD: &str = "fixture:extension-tail-covered-head";
        const STAGE: &str = "extension-tail-component";
        const TAIL: u64 = 3;
        let stream = format!("{OUTBOX_STREAM_PREFIX}7");
        let row = OutboxRow::AckAdvanced {
            source_log_sequence: 1,
            participant_id: 7,
            through_seq: 0,
        };
        let payload = encode_row(&row)?;
        let payload_bytes = u64::try_from(payload.len())?;
        let pair = history_pair("extension-tail-cursor-v1", |store, range| {
            let old_head = block_on(store.read_value(HEAD))??.unwrap_or_default();
            let existing = if old_head == 0 { 0 } else { old_head + TAIL };
            let new_head = range.end;
            let target = new_head
                .checked_add(TAIL)
                .ok_or("extension fixture head overflow")?;
            let count = usize::try_from(
                target
                    .checked_sub(existing)
                    .ok_or("extension fixture head regressed")?,
            )?;
            assert_eq!(
                block_on(store.append_all(&stream, vec![payload.clone(); count], existing))??,
                target
            );
            block_on(store.cas(HEAD, old_head, new_head))??;
            Ok(())
        })?;
        let expected: StageReads = BTreeMap::from([(
            STAGE.to_owned(),
            BTreeMap::from([
                (
                    ReadSource::Value(HEAD.to_owned()),
                    ReadCounts { rows: 1, bytes: 8 },
                ),
                (
                    ReadSource::Stream(stream),
                    ReadCounts {
                        rows: TAIL,
                        bytes: TAIL * payload_bytes,
                    },
                ),
            ]),
        )]);
        let measured = pair.assert_stage(STAGE, &expected, |store| {
            during_stage(STAGE, || {
                let head =
                    block_on(store.read_value(HEAD))??.ok_or("extension fixture head absent")?;
                let mut saved = positions();
                saved.extension_sequence = head;
                saved.merged_extension_sequence = head;
                let checkpoint = RestoreCheckpoint::new(saved, [9; 32], Live(7))?;
                let start = TailStart::from_checkpoint(&checkpoint);
                let log = OutboxLog::new(store, 7);
                let mut cursor = log.restore_cursor_from(start.extension_sequence());
                let mut sequences = Vec::new();
                block_on(async {
                    while cursor.front().await?.is_some() {
                        let (sequence, actual) =
                            cursor.pop_front().ok_or("extension cursor lost row")?;
                        assert_eq!(actual, row);
                        sequences.push(sequence);
                    }
                    Ok::<(), Box<dyn Error>>(())
                })??;
                assert_eq!(sequences, (head..head + TAIL).collect::<Vec<_>>());
                assert_eq!(cursor.confirmed_head(), Some(head + TAIL));
                Ok(())
            })
        })?;
        println!("extension tail measured reads: {measured:?}");
        Ok(())
    }
}

#[cfg(test)]
mod metadata_tests {
    use super::super::barrier::ReceiptCapacityLimits;
    use super::super::outbox::{ConversationOutbox, ConversationOutboxLimits};
    use super::super::restore_checkpoint_history::{CheckpointHistoryRoot, ROOT_ENCODED_BYTES};
    use super::super::restore_checkpoint_locator::CheckpointBodyKey;
    use super::super::state::ConversationAuthority;
    use super::super::state::checkpoint_owner::ConversationAuthorityRestoreContext;
    use super::*;
    use liminal_protocol::lifecycle::{
        ConversationDecision, ConversationGenesis, ParticipantConversation,
    };
    use std::error::Error;

    fn committed_genesis() -> Result<ConversationAuthority, Box<dyn Error>> {
        let config = crate::server::participant::test_participant_config();
        let mut owner =
            ConversationAuthority::empty(7, ReceiptCapacityLimits::from_config(&config));
        let fresh = ParticipantConversation::from_genesis(ConversationGenesis::new(7));
        let ConversationDecision::Commit(commit) = fresh.decide_genesis_validation() else {
            return Err("genesis validation refused".into());
        };
        owner.shell = Some(commit.commit());
        owner.next_log_sequence = 1;
        let limits = ConversationOutboxLimits::try_new(
            config.max_retained_record_rows,
            config.identity_slots,
        )?;
        owner.outbox = Some(ConversationOutbox::restore(7, Vec::new(), limits)?);
        Ok(owner)
    }

    #[test]
    fn prepared_restore_retains_saved_root_audit_and_prefix() -> Result<(), Box<dyn Error>> {
        let config = crate::server::participant::test_participant_config();
        let root = CheckpointHistoryRoot::empty(7);
        let audit = ComposedTerminalValidation::for_identity_slots(config.identity_slots)?;
        let checkpoint = prepare_replayed_checkpoint(
            committed_genesis()?,
            &audit,
            CheckpointSchema::V4Suffix,
            root,
            [7; 32],
        )?;
        let positions = checkpoint.positions();
        let bytes = checkpoint.encode_canonical()?;
        let locator = CheckpointLocator::new(
            7,
            positions.covered_log_sequence,
            positions.extension_sequence,
            [7; 32],
            CheckpointBodyKey::for_body(&bytes),
        )?;
        let context =
            ConversationAuthorityRestoreContext::new(positions, &root.encode_canonical(), &config)?;
        let restored =
            RestoreCheckpoint::<ConversationAuthority, ROOT_ENCODED_BYTES>::decode_canonical(
                &bytes, 7, &context,
            )?;
        let prepared = prepare_checkpoint_restore(restored, &locator, &config)?;
        assert_eq!(prepared.root, root);
        assert_eq!(
            prepared.start.operation_sequence(),
            positions.next_log_sequence
        );
        assert_eq!(
            prepared.start.extension_sequence(),
            positions.extension_sequence
        );
        assert_eq!(prepared.prefix_digest, [7; 32]);
        assert_eq!(
            prepared.audit.encode_checkpoint()?,
            audit.encode_checkpoint()?
        );
        let metadata = prepared.owner.checkpoint_metadata()?;
        assert_eq!(metadata.history_root(), &root.encode_canonical());
        assert_eq!(metadata.prefix_digest(), [7; 32]);
        let recaptured = prepare_replayed_checkpoint(
            prepared.owner,
            &prepared.audit,
            positions.schema,
            prepared.root,
            prepared.prefix_digest,
        )?;
        assert_eq!(recaptured.encode_canonical()?, bytes);
        Ok(())
    }

    #[test]
    fn replayed_owner_refuses_a_foreign_history_root() -> Result<(), Box<dyn Error>> {
        let config = crate::server::participant::test_participant_config();
        let audit = ComposedTerminalValidation::for_identity_slots(config.identity_slots)?;
        assert!(
            prepare_replayed_checkpoint(
                committed_genesis()?,
                &audit,
                CheckpointSchema::V4Suffix,
                CheckpointHistoryRoot::empty(8),
                [7; 32],
            )
            .is_err()
        );
        Ok(())
    }
}