liminal-protocol 0.14.1

//! Restores executable frontier ownership only from a validated lifecycle capsule.

use alloc::vec::Vec;

use crate::algebra::{ResourceVector, WideResourceVector};
use crate::lifecycle::{
    ClosureAccounting, OrdinaryProjectionError, ParticipantConversation, RestoredLifecycleParts,
    RestoredParticipantLifecycle, RetainedRecordCharge,
};
use crate::wire::CredentialDigest;

use super::LiveFrontierOwner;

/// Participant values moved from the validated credential-digest lifecycle capsule.
pub type CheckpointParticipant = RestoredParticipantLifecycle<
    CredentialDigest,
    CredentialDigest,
    CredentialDigest,
    CredentialDigest,
>;

/// The original shell and participants beside their restored executable frontier.
#[derive(Debug)]
#[non_exhaustive]
pub struct RestoredLiveFrontierOwner {
    /// The capsule's original conversation shell.
    pub conversation: ParticipantConversation,
    /// The capsule's original participant values and tombstones.
    pub participants: Vec<CheckpointParticipant>,
    /// The coupled frontier, accounting and exact keyed retained charges.
    pub owner: LiveFrontierOwner,
    /// The exact saved cursor facts, present only for owed closure.
    pub cursor_episode: Option<crate::lifecycle::NonzeroDebtCursorEpisode>,
    pub(in crate::lifecycle) observer_progress: u64,
}

/// Named checkpoint omissions and joint-state validation failures.
#[derive(Debug, PartialEq, Eq)]
pub enum LiveFrontierRestoreError {
    /// The checkpoint contains no enrolled lifecycle.
    MissingLifecycle,
    /// The checkpoint omits closure accounting.
    MissingClosureAccounting,
    /// The checkpoint omits keyed retained charges.
    MissingRetainedCharges,
    /// Owed closure omits its saved cursor episode.
    MissingCursorEpisode,
    /// Clear closure unexpectedly carries a cursor episode.
    UnexpectedCursorEpisode,
    /// The original accounting reader refused the source.
    AccountingCodec(crate::lifecycle::FrontierAccountingDecodeError),
    /// The original retained-charge reader refused the source.
    RetainedChargeCodec(crate::lifecycle::CapsuleDecodeError),
    /// The original joint cursor-episode validator refused the source.
    CursorEpisode(crate::lifecycle::StorageRestoreError),
    /// The original saved-episode writer refused canonical encoding.
    CursorEpisodeEncoding(crate::lifecycle::CursorFactEncodeError),
    /// The shell and restored frontier belong to different conversations.
    ForeignConversation,
    /// The capsule's closure and decoded accounting disagree.
    ClosureMismatch,
    /// Retained rows exceed their configured limit.
    RetainedRecordLimit,
    /// Native retained-row, marker or baseline validation refused.
    Accounting(OrdinaryProjectionError),
}

impl LiveFrontierOwner {
    /// Restores coupled ownership from the validated capsule and decoded saved values.
    ///
    /// No participant, frontier, closure, charge or accounting field is regenerated.
    ///
    /// # Errors
    ///
    /// Names every absent saved field and refuses inconsistent closure, retained keys,
    /// marker ledgers, baseline or retention limit.
    fn restore_checkpoint_parts(
        parts: RestoredLifecycleParts,
        accounting: ClosureAccounting,
        charges: Vec<RetainedRecordCharge>,
        cursor_episode: Option<crate::lifecycle::NonzeroDebtCursorEpisode>,
        saved: FrontierCheckpointSource<'_>,
    ) -> Result<RestoredLiveFrontierOwner, LiveFrontierRestoreError> {
        let lifecycle = parts
            .lifecycle
            .ok_or(LiveFrontierRestoreError::MissingLifecycle)?;
        if lifecycle.frontiers().conversation_id() != parts.conversation.conversation_id() {
            return Err(LiveFrontierRestoreError::ForeignConversation);
        }
        if lifecycle.closure() != accounting.state() {
            return Err(LiveFrontierRestoreError::ClosureMismatch);
        }
        // A drained binding terminal is retained above the cap it pended at
        // (`LiveFrontierOwner::drain_pending_terminal`), by at most the
        // identity-slot candidate bound; the frontier restore admits the
        // same overage (11 October 2026).
        if lifecycle.frontiers().retained_records().len() as u128
            > u128::from(saved.retained_record_limit)
                .saturating_add(u128::from(lifecycle.frontiers().identity_slot_limit()))
        {
            return Err(LiveFrontierRestoreError::RetainedRecordLimit);
        }
        validate_saved_accounting(
            lifecycle.frontiers(),
            &accounting,
            &charges,
            saved.marker_max,
        )
        .map_err(LiveFrontierRestoreError::Accounting)?;
        let (participants, frontiers, _) = lifecycle.into_parts();
        Ok(RestoredLiveFrontierOwner {
            conversation: parts.conversation,
            participants,
            cursor_episode,
            observer_progress: saved.observer_progress,
            owner: Self {
                frontiers,
                closure_accounting: accounting,
                retained_charges: charges,
                retained_record_limit: saved.retained_record_limit,
            },
        })
    }
}

fn validate_saved_accounting(
    frontiers: &crate::lifecycle::ClaimFrontiers,
    accounting: &ClosureAccounting,
    charges: &[RetainedRecordCharge],
    marker_max: ResourceVector,
) -> Result<(), OrdinaryProjectionError> {
    let records = frontiers.retained_records();
    let high_end = u128::from(frontiers.sequence().ledger().high_watermark()) + 1;
    if high_end.checked_sub(frontiers.retained_floor()) != Some(records.len() as u128)
        || records.len() != charges.len()
    {
        return Err(OrdinaryProjectionError::RetainedSuffix);
    }
    // A credited marker is charged at `marker_max`, exactly as the live
    // admission projection charges it (`retained_charge_at_floor`): its
    // credit holds the whole reservation however small the marker record
    // is. Summing the exact charge here instead refused every checkpoint of
    // a conversation holding a marker smaller than the maximum (the door
    // room, 11 October 2026: stored 94,481,201 bytes, derived 94,458,959),
    // and every boot replayed the room from row 0.
    let credited: Vec<_> = frontiers
        .retained_marker_records()
        .iter()
        .map(|record| record.delivery_seq)
        .collect();
    let mut retained = WideResourceVector::default();
    for (index, (record, charge)) in records.iter().zip(charges).enumerate() {
        if u128::from(record.delivery_seq) != frontiers.retained_floor() + index as u128
            || charge.delivery_seq() != record.delivery_seq
            || charge.admission_order() != record.admission_order
        {
            return Err(OrdinaryProjectionError::RetainedChargeKey {
                index: u64::try_from(index).map_err(|_| OrdinaryProjectionError::RetainedSuffix)?,
            });
        }
        let resource = charge.encoded_charge();
        if resource.entries != 1 {
            return Err(OrdinaryProjectionError::RetainedEntryCharge {
                delivery_seq: charge.delivery_seq(),
                entries: resource.entries,
            });
        }
        let resource = if credited.contains(&record.delivery_seq) {
            marker_max
        } else {
            resource
        };
        retained =
            super::super::ordinary_record_projection::checked_add_resource(retained, resource)?;
    }
    let pending = frontiers.pending_marker_candidate_count();
    let retained_credits = u64::try_from(frontiers.retained_marker_records().len())
        .map_err(|_| OrdinaryProjectionError::MarkerCredit)?;
    let credits = retained_credits
        .checked_add(pending)
        .ok_or(OrdinaryProjectionError::MarkerCredit)?;
    if credits != accounting.marker_capacity_credits() {
        return Err(OrdinaryProjectionError::MarkerCreditAccounting {
            derived: credits,
            stored: accounting.marker_capacity_credits(),
        });
    }
    let anchors = frontiers
        .unaccepted_marker_anchor_count()
        .checked_add(pending)
        .ok_or(OrdinaryProjectionError::MarkerAnchor)?;
    if anchors != accounting.marker_anchors() {
        return Err(OrdinaryProjectionError::MarkerAnchorAccounting {
            derived: anchors,
            stored: accounting.marker_anchors(),
        });
    }
    let baseline = super::super::ordinary_record_projection::retained_baseline_wide(
        retained,
        frontiers.identity_slot_limit(),
        retained_credits,
        marker_max,
    )?;
    if baseline != accounting.baseline() {
        return Err(OrdinaryProjectionError::BaselineAccounting {
            derived: baseline,
            stored: accounting.baseline(),
        });
    }
    Ok(())
}

impl LiveFrontierOwner {
    /// Decodes every saved coupled field against a validated lifecycle capsule.
    ///
    /// # Errors
    ///
    /// Names absent fields, retains the original codec failures, and rejects
    /// mismatched cursor facts, keyed charges, closure, marker ledgers or baseline.
    pub fn restore_checkpoint(
        parts: RestoredLifecycleParts,
        saved: FrontierCheckpointSource<'_>,
    ) -> Result<RestoredLiveFrontierOwner, LiveFrontierRestoreError> {
        let lifecycle = parts
            .lifecycle
            .as_ref()
            .ok_or(LiveFrontierRestoreError::MissingLifecycle)?;
        let accounting = Self::decode_closure_accounting_source(
            lifecycle.closure(),
            saved
                .closure_accounting
                .ok_or(LiveFrontierRestoreError::MissingClosureAccounting)?,
        )
        .map_err(LiveFrontierRestoreError::AccountingCodec)?;
        let charges = Self::decode_retained_charge_source(
            saved
                .retained_charges
                .ok_or(LiveFrontierRestoreError::MissingRetainedCharges)?,
        )
        .map_err(LiveFrontierRestoreError::RetainedChargeCodec)?;
        let episode = match lifecycle.closure() {
            crate::lifecycle::ClosureState::Clear => {
                if saved.cursor_episode.is_some() {
                    return Err(LiveFrontierRestoreError::UnexpectedCursorEpisode);
                }
                None
            }
            crate::lifecycle::ClosureState::Owed { debt, .. } => {
                let bytes = saved
                    .cursor_episode
                    .ok_or(LiveFrontierRestoreError::MissingCursorEpisode)?;
                Some(
                    crate::lifecycle::NonzeroDebtCursorEpisode::decode_checkpoint(
                        bytes,
                        lifecycle.frontiers(),
                        debt,
                        saved.observer_progress,
                    )
                    .map_err(LiveFrontierRestoreError::CursorEpisode)?,
                )
            }
        };
        Self::restore_checkpoint_parts(parts, accounting, charges, episode, saved)
    }
}

/// Borrowed canonical saved fields and their required configured bounds.
#[derive(Clone, Copy, Debug)]
pub struct FrontierCheckpointSource<'a> {
    /// Canonical closure-accounting scalar bytes; absence is a named refusal.
    pub closure_accounting: Option<&'a [u8]>,
    /// Canonical keyed retained-charge bytes; absence is a named refusal.
    pub retained_charges: Option<&'a [u8]>,
    /// Exact saved cursor episode, required for owed closure and absent for clear.
    pub cursor_episode: Option<&'a [u8]>,
    /// Saved observer progress used to validate the episode.
    pub observer_progress: u64,
    /// The configured bound on retained physical rows.
    pub retained_record_limit: u64,
    /// The configured maximum marker charge used by native baseline accounting.
    pub marker_max: ResourceVector,
}