liminal_protocol/lifecycle/

checkpoint.rs

//! Lifecycle checkpoint capsule: every protocol-owned value of one
//! conversation's lifecycle at a log position, in canonical bytes.
//!
//! A storage binding lends its live values to [`LifecycleCapsule::capture`]
//! through [`LifecycleView`], which holds references only, so a binding can
//! lend what it has but never fabricate a value. The capsule's bytes are this
//! module's own stable v1 `LPCK` envelope; the binding stores them beside its
//! own state and never reads their fields. [`LifecycleCapsule::restore`] is
//! the only way from a capsule back to live values, and it validates the
//! capsule as a whole first.
//!
//! The capsule cannot be built by a struct literal:
//!
//! ```compile_fail
//! use liminal_protocol::lifecycle::LifecycleCapsule;
//!
//! let _ = LifecycleCapsule { conversation_id: 1, covered_log_sequence: 0 };
//! ```

use alloc::collections::BTreeSet;
use alloc::vec::Vec;

use crate::wire::{ConversationId, ParticipantId};

use super::checkpoint_codec::{
    CLOSURE_SECTION_TAG, DigestDescription, FRONTIERS_SECTION_TAG, PARTICIPANTS_SECTION_TAG,
    SEALED_SECTION_TAG, decode_description, decode_sealed_description, encode_description,
    encode_sealed_description,
};
use super::claim_frontier::ClaimFrontiers;
use super::conversation::{ConversationGenesis, ParticipantConversation};
use super::edge::ClosureState;
use super::membership::RetiredIdentity;
use super::storage::{
    ConversationDescribeError, ConversationStateRestoreError, ParticipantConversationState,
    ParticipantLifecycleRestore, ParticipantLifecycleView, StorageRestoreError,
    describe_conversation,
};

/// A participant view over the 32-byte digests every server stores.
pub type DigestParticipantView<'a> =
    ParticipantLifecycleView<'a, [u8; 32], [u8; 32], [u8; 32], [u8; 32]>;

/// Whole-conversation state over the 32-byte digests every server stores.
pub type DigestConversationState =
    ParticipantConversationState<[u8; 32], [u8; 32], [u8; 32], [u8; 32]>;

/// One retired identity over the 32-byte digests every server stores.
pub type DigestRetiredIdentity = RetiredIdentity<[u8; 32], [u8; 32], [u8; 32]>;

type DigestParticipantRestore = ParticipantLifecycleRestore<
    [u8; DIGEST_LEN],
    [u8; DIGEST_LEN],
    [u8; DIGEST_LEN],
    [u8; DIGEST_LEN],
>;

const CAPSULE_MAGIC: [u8; 4] = *b"LPCK";
const CAPSULE_CODEC_MAJOR: u16 = 1;
/// Minor 0: the conversation section, then the three lifecycle sections or
/// none. Every capsule written before the seal section decodes as minor 0.
const CAPSULE_CODEC_MINOR: u16 = 0;
/// Minor 1: the conversation section, the participants section and the
/// sealed section. Written only for a sealed conversation.
const CAPSULE_CODEC_MINOR_SEALED: u16 = 1;
/// Magic, major, minor, conversation id, covered log sequence, body length.
const CAPSULE_HEADER_LEN: usize = 28;
/// Section tag and section body length.
const SECTION_HEADER_LEN: u32 = 6;
const CONVERSATION_SECTION_TAG: u16 = 1;
/// Next event ordinal and flags.
const CONVERSATION_SECTION_LEN: u32 = 9;
const GENESIS_VALIDATED_FLAG: u8 = 0b1;
/// The conversation section with its header, which every capsule carries.
const CONVERSATION_FRAME_LEN: u32 = SECTION_HEADER_LEN + CONVERSATION_SECTION_LEN;
/// Every assigned section tag. The conversation section is always present;
/// the participants, frontiers and closure sections follow it in this order,
/// all three or none, or, for a sealed conversation, the participants section
/// and the sealed section.
const SECTION_TAGS: [u16; 5] = [
    CONVERSATION_SECTION_TAG,
    PARTICIPANTS_SECTION_TAG,
    FRONTIERS_SECTION_TAG,
    CLOSURE_SECTION_TAG,
    SEALED_SECTION_TAG,
];
/// Width of every generic digest the capsule carries: the enrollment and
/// Leave fingerprints and both request verifiers.
const DIGEST_LEN: usize = 32;

/// What a storage binding lends to [`LifecycleCapsule::capture`]: references
/// to the protocol values it holds, and the log position they cover.
#[derive(Debug)]
pub struct LifecycleView<'a> {
    /// The conversation's event-sourced shell.
    pub conversation: &'a ParticipantConversation,
    /// The participants, frontiers and closure, once the conversation has
    /// any; `None` before its first enrollment and after its seal.
    pub lifecycle: Option<ConversationLifecycleView<'a>>,
    /// The retired identities and the final ledger watermark, once the
    /// conversation is sealed; `None` otherwise, and never beside `lifecycle`.
    pub sealed: Option<SealedLifecycleView<'a>>,
    /// The last log sequence whose effects these values include.
    pub covered_log_sequence: u64,
}

/// What a sealed conversation lends: the seal retired its frontier owner, and
/// the observer sources it still covers need its final sequence-ledger high
/// watermark beside the retired generations.
#[derive(Debug)]
pub struct SealedLifecycleView<'a> {
    /// Every retired participant, strictly ascending by participant id; a
    /// seal leaves no live slot.
    pub participants: Vec<DigestParticipantView<'a>>,
    /// The retired sequence ledger's high watermark at the seal.
    pub ledger_high_watermark: u64,
}

/// The lifecycle values of one enrolled conversation, lent by reference.
#[derive(Debug)]
pub struct ConversationLifecycleView<'a> {
    /// Every live and retired participant, strictly ascending by participant
    /// id.
    pub participants: Vec<DigestParticipantView<'a>>,
    /// Both coupled claim frontiers.
    pub frontiers: &'a ClaimFrontiers,
    /// The signed cap on retained record facts the frontier owner keeps.
    pub retained_record_limit: u64,
    /// The current closure debt and edge.
    pub closure: &'a ClosureState,
}

/// Every protocol-owned value of one conversation's lifecycle at a log
/// position. Built only by [`LifecycleCapsule::capture`] or
/// [`LifecycleCapsule::decode_canonical`].
#[derive(Debug, PartialEq, Eq)]
pub struct LifecycleCapsule {
    conversation_id: ConversationId,
    covered_log_sequence: u64,
    conversation: ConversationSection,
    lifecycle: Option<LifecycleSections>,
    sealed: Option<SealedSections>,
    /// Exact body length after the fixed header, proven to fit at capture or
    /// read from the header at decode.
    body_len: u32,
}

/// The described retired participants and the sealed watermark, beside their
/// exact canonical section bytes (headers included).
#[derive(Debug, PartialEq, Eq)]
struct SealedSections {
    participants: Vec<DigestParticipantRestore>,
    ledger_high_watermark: u64,
    encoded: Vec<u8>,
}

/// The conversation shell's state: its genesis is the conversation id.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
struct ConversationSection {
    next_event_ordinal: u64,
    genesis_validated: bool,
}

impl ConversationSection {
    /// Genesis validation is the first event and happens once, so a shell
    /// past ordinal 0 has validated its genesis and one at ordinal 0 has not.
    const fn is_consistent(self) -> bool {
        self.genesis_validated == (self.next_event_ordinal > 0)
    }
}

/// The described participants, frontiers and closure, beside their exact
/// canonical section bytes (headers included).
#[derive(Debug, PartialEq, Eq)]
struct LifecycleSections {
    description: DigestDescription<DIGEST_LEN>,
    encoded: Vec<u8>,
}

impl LifecycleCapsule {
    /// Captures the lent values. Nothing is consumed; generic digests are
    /// copied out of the borrowed values.
    ///
    /// # Errors
    ///
    /// Returns [`CapsuleCaptureError`] when the frontiers belong to another
    /// conversation, the participants are not strictly ascending, the live
    /// closure cannot be described, or a count or length does not fit `u32`.
    pub fn capture(view: &LifecycleView<'_>) -> Result<Self, CapsuleCaptureError> {
        let conversation_id = view.conversation.conversation_id();
        if view.lifecycle.is_some() && view.sealed.is_some() {
            return Err(CapsuleCaptureError::SealedBesideLifecycle);
        }
        let lifecycle = view
            .lifecycle
            .as_ref()
            .map(|lifecycle| capture_lifecycle(conversation_id, lifecycle))
            .transpose()?;
        let sealed = view.sealed.as_ref().map(capture_sealed).transpose()?;
        let lifecycle_len = lifecycle
            .as_ref()
            .map_or(0, |sections| sections.encoded.len())
            .checked_add(sealed.as_ref().map_or(0, |sections| sections.encoded.len()))
            .ok_or(CapsuleCaptureError::LengthOverflow)?;
        let body_len = u32::try_from(lifecycle_len)
            .map_err(|_| CapsuleCaptureError::LengthOverflow)?
            .checked_add(CONVERSATION_FRAME_LEN)
            .ok_or(CapsuleCaptureError::LengthOverflow)?;
        Ok(Self {
            conversation_id,
            covered_log_sequence: view.covered_log_sequence,
            conversation: ConversationSection {
                next_event_ordinal: view.conversation.next_event_ordinal(),
                genesis_validated: view.conversation.genesis_validated(),
            },
            lifecycle,
            sealed,
            body_len,
        })
    }

    /// Whether the capsule carries a sealed conversation.
    #[must_use]
    pub const fn is_sealed(&self) -> bool {
        self.sealed.is_some()
    }

    /// The conversation these values belong to.
    #[must_use]
    pub const fn conversation_id(&self) -> ConversationId {
        self.conversation_id
    }

    /// The last log sequence whose effects the capsule includes.
    #[must_use]
    pub const fn covered_log_sequence(&self) -> u64 {
        self.covered_log_sequence
    }

    /// Encodes the stable v1 capsule envelope in network byte order.
    #[must_use]
    pub fn encode_canonical(&self) -> Vec<u8> {
        let mut encoded = Vec::with_capacity(CAPSULE_HEADER_LEN + self.body_len as usize);
        let minor = if self.sealed.is_some() {
            CAPSULE_CODEC_MINOR_SEALED
        } else {
            CAPSULE_CODEC_MINOR
        };
        encoded.extend_from_slice(&CAPSULE_MAGIC);
        encoded.extend_from_slice(&CAPSULE_CODEC_MAJOR.to_be_bytes());
        encoded.extend_from_slice(&minor.to_be_bytes());
        encoded.extend_from_slice(&self.conversation_id.to_be_bytes());
        encoded.extend_from_slice(&self.covered_log_sequence.to_be_bytes());
        encoded.extend_from_slice(&self.body_len.to_be_bytes());
        encoded.extend_from_slice(&CONVERSATION_SECTION_TAG.to_be_bytes());
        encoded.extend_from_slice(&CONVERSATION_SECTION_LEN.to_be_bytes());
        encoded.extend_from_slice(&self.conversation.next_event_ordinal.to_be_bytes());
        encoded.push(if self.conversation.genesis_validated {
            GENESIS_VALIDATED_FLAG
        } else {
            0
        });
        if let Some(sections) = &self.lifecycle {
            encoded.extend_from_slice(&sections.encoded);
        }
        if let Some(sections) = &self.sealed {
            encoded.extend_from_slice(&sections.encoded);
        }
        encoded
    }

    /// Decodes one exact canonical v1 capsule.
    ///
    /// # Errors
    ///
    /// Returns [`CapsuleDecodeError`] for truncation, an invalid magic prefix,
    /// an unsupported codec version, a declared length that differs from the
    /// supplied bytes, a section out of order, missing, unknown or of the wrong
    /// length, a section whose fields break their invariants, participants out
    /// of order, or frontiers that belong to another conversation.
    pub fn decode_canonical(input: &[u8]) -> Result<Self, CapsuleDecodeError> {
        if input.len() < CAPSULE_HEADER_LEN {
            return Err(CapsuleDecodeError::Truncated {
                required: CAPSULE_HEADER_LEN,
                available: input.len(),
            });
        }
        if input.get(0..4) != Some(CAPSULE_MAGIC.as_slice()) {
            return Err(CapsuleDecodeError::InvalidMagic);
        }
        let major = take_u16(input, 4)?;
        let minor = take_u16(input, 6)?;
        if major != CAPSULE_CODEC_MAJOR || minor > CAPSULE_CODEC_MINOR_SEALED {
            return Err(CapsuleDecodeError::UnsupportedCodec { major, minor });
        }
        let conversation_id = take_u64(input, 8)?;
        let covered_log_sequence = take_u64(input, 16)?;
        let declared_body_len = take_u32(input, 24)?;
        let actual_body_len = input.len() - CAPSULE_HEADER_LEN;
        if usize::try_from(declared_body_len).map_err(|_| CapsuleDecodeError::LengthOverflow)?
            != actual_body_len
        {
            return Err(CapsuleDecodeError::NonCanonicalLength {
                declared_body_len,
                actual_body_len,
            });
        }
        let mut at = CAPSULE_HEADER_LEN;
        let conversation =
            decode_conversation(take_section(input, &mut at, CONVERSATION_SECTION_TAG)?)?;
        let (lifecycle, sealed) = if input.len() - at < SECTION_HEADER_LEN as usize {
            (None, None)
        } else {
            let start = at;
            let participants = take_section(input, &mut at, PARTICIPANTS_SECTION_TAG)?;
            let next_tag = if input.len() - at < SECTION_HEADER_LEN as usize {
                None
            } else {
                Some(take_u16(input, at)?)
            };
            if next_tag == Some(SEALED_SECTION_TAG) {
                if minor != CAPSULE_CODEC_MINOR_SEALED {
                    return Err(CapsuleDecodeError::UnknownSection {
                        tag: SEALED_SECTION_TAG,
                    });
                }
                let sealed = take_section(input, &mut at, SEALED_SECTION_TAG)?;
                let (participants, ledger_high_watermark) =
                    decode_sealed_description::<DIGEST_LEN>(participants, sealed)?;
                let encoded = input
                    .get(start..at)
                    .ok_or(CapsuleDecodeError::LengthOverflow)?
                    .to_vec();
                (
                    None,
                    Some(SealedSections {
                        participants,
                        ledger_high_watermark,
                        encoded,
                    }),
                )
            } else {
                (
                    Some(decode_lifecycle(
                        input,
                        &mut at,
                        start,
                        participants,
                        conversation_id,
                    )?),
                    None,
                )
            }
        };
        if minor == CAPSULE_CODEC_MINOR_SEALED && sealed.is_none() {
            return Err(CapsuleDecodeError::MissingSection {
                tag: SEALED_SECTION_TAG,
            });
        }
        if at != input.len() {
            return Err(CapsuleDecodeError::TrailingBytes {
                trailing: input.len() - at,
            });
        }
        Ok(Self {
            conversation_id,
            covered_log_sequence,
            conversation,
            lifecycle,
            sealed,
            body_len: declared_body_len,
        })
    }

    /// Associates exact origin sources before restoring the complete capsule.
    ///
    /// # Errors
    ///
    /// Refuses malformed, unordered, missing or foreign origin sources, then
    /// propagates the existing joint lifecycle validation failure.
    #[doc(hidden)]
    pub fn restore_with_binding_sources(
        self,
        sources: &[(ParticipantId, &[u8])],
    ) -> Result<RestoredLifecycle, BindingOriginSourceError> {
        let mut sources = sources.iter();
        if let Some(lifecycle) = &self.lifecycle {
            for participant in &lifecycle.description.restore.participants {
                if let super::storage::ParticipantLifecycleRestore::Live {
                    identity,
                    binding_origin: Some(expected),
                    ..
                } = participant
                {
                    let Some(&(participant_id, bytes)) = sources.next() else {
                        return Err(BindingOriginSourceError::Association {
                            participant_id: identity.participant_id,
                        });
                    };
                    let origin = super::checkpoint_codec::decode_binding_origin_source(bytes)
                        .map_err(|error| BindingOriginSourceError::Decode {
                            participant_id,
                            error,
                        })?;
                    if participant_id != identity.participant_id || origin != *expected {
                        return Err(BindingOriginSourceError::Association { participant_id });
                    }
                }
            }
        }
        if let Some(&(participant_id, _)) = sources.next() {
            return Err(BindingOriginSourceError::Association { participant_id });
        }
        self.restore().map_err(BindingOriginSourceError::Restore)
    }

    /// Validates the capsule as a whole and hands back the live values.
    ///
    /// The lifecycle goes through the one joint participant-conversation
    /// validation, through its crate-private entry that also checks and
    /// installs the fenced-marker issuance latch the capsule carried: a latch
    /// naming a take the restored state could not have granted is refused.
    ///
    /// # Errors
    ///
    /// Returns [`CapsuleRestoreError`] when the capsule's sections are
    /// inconsistent with each other or with their own invariants.
    pub fn restore(self) -> Result<RestoredLifecycle, CapsuleRestoreError> {
        if !self.conversation.is_consistent() {
            return Err(CapsuleRestoreError::InconsistentConversation);
        }
        let lifecycle = self
            .lifecycle
            .map(|sections| sections.description.into_state())
            .transpose()
            .map_err(CapsuleRestoreError::Lifecycle)?;
        let sealed = self.sealed.map(restore_sealed).transpose()?;
        Ok(RestoredLifecycle {
            covered_log_sequence: self.covered_log_sequence,
            restored_fenced_sources: BTreeSet::new(),
            conversation: ParticipantConversation::from_checkpoint(
                ConversationGenesis::new(self.conversation_id),
                self.conversation.next_event_ordinal,
                self.conversation.genesis_validated,
            ),
            lifecycle,
            sealed,
        })
    }
}

/// Describes and encodes one lent lifecycle, refusing a view the capsule
/// could not restore as the same conversation.
fn capture_lifecycle(
    conversation_id: ConversationId,
    view: &ConversationLifecycleView<'_>,
) -> Result<LifecycleSections, CapsuleCaptureError> {
    if view.frontiers.conversation_id() != conversation_id {
        return Err(CapsuleCaptureError::ForeignConversation);
    }
    for pair in view.participants.windows(2) {
        if let [earlier, later] = pair
            && earlier.participant_id() >= later.participant_id()
        {
            return Err(CapsuleCaptureError::ParticipantOutOfOrder {
                participant_id: later.participant_id(),
            });
        }
    }
    let description = describe_conversation(
        &view.participants,
        view.frontiers,
        view.retained_record_limit,
        *view.closure,
    )
    .map_err(CapsuleCaptureError::Describe)?;
    let encoded = encode_description(&description).ok_or(CapsuleCaptureError::LengthOverflow)?;
    Ok(LifecycleSections {
        description,
        encoded,
    })
}

/// Describes and encodes one lent sealed conversation, refusing a live
/// participant or an unordered one.
fn capture_sealed(view: &SealedLifecycleView<'_>) -> Result<SealedSections, CapsuleCaptureError> {
    for pair in view.participants.windows(2) {
        if let [earlier, later] = pair
            && earlier.participant_id() >= later.participant_id()
        {
            return Err(CapsuleCaptureError::ParticipantOutOfOrder {
                participant_id: later.participant_id(),
            });
        }
    }
    if let Some(live) = view
        .participants
        .iter()
        .find(|participant| matches!(participant, DigestParticipantView::Live { .. }))
    {
        return Err(CapsuleCaptureError::LiveParticipantSealed {
            participant_id: live.participant_id(),
        });
    }
    let participants: Vec<_> = view
        .participants
        .iter()
        .map(ParticipantLifecycleView::storage_description)
        .collect();
    let encoded = encode_sealed_description(&participants, view.ledger_high_watermark)
        .ok_or(CapsuleCaptureError::LengthOverflow)?;
    Ok(SealedSections {
        participants,
        ledger_high_watermark: view.ledger_high_watermark,
        encoded,
    })
}

/// Restores every retired identity a sealed capsule carries.
fn restore_sealed(sections: SealedSections) -> Result<SealedLifecycle, CapsuleRestoreError> {
    let mut retired = Vec::with_capacity(sections.participants.len());
    for participant in sections.participants {
        match participant {
            ParticipantLifecycleRestore::Retired(identity) => {
                retired.push(identity.restore().map_err(CapsuleRestoreError::Sealed)?);
            }
            ParticipantLifecycleRestore::Live { identity, .. } => {
                return Err(CapsuleRestoreError::SealedLiveParticipant {
                    participant_id: identity.participant_id,
                });
            }
        }
    }
    Ok(SealedLifecycle {
        retired,
        ledger_high_watermark: sections.ledger_high_watermark,
    })
}

/// Decodes the frontiers and closure sections, which follow the participants
/// section `participants` (taken from `start`) all together.
fn decode_lifecycle(
    input: &[u8],
    at: &mut usize,
    start: usize,
    participants: &[u8],
    conversation_id: ConversationId,
) -> Result<LifecycleSections, CapsuleDecodeError> {
    let frontiers = take_section(input, at, FRONTIERS_SECTION_TAG)?;
    let closure = take_section(input, at, CLOSURE_SECTION_TAG)?;
    let description = decode_description::<DIGEST_LEN>(participants, frontiers, closure)?;
    if description.restore.frontiers.conversation_id != conversation_id {
        return Err(CapsuleDecodeError::ForeignConversation {
            tag: FRONTIERS_SECTION_TAG,
        });
    }
    let encoded = input
        .get(start..*at)
        .ok_or(CapsuleDecodeError::LengthOverflow)?
        .to_vec();
    Ok(LifecycleSections {
        description,
        encoded,
    })
}

/// Takes the next section, which must carry `expected`, and moves past it.
fn take_section<'a>(
    input: &'a [u8],
    at: &mut usize,
    expected: u16,
) -> Result<&'a [u8], CapsuleDecodeError> {
    if *at == input.len() {
        return Err(CapsuleDecodeError::MissingSection { tag: expected });
    }
    let tag = take_u16(input, *at)?;
    let len = take_u32(input, *at + 2)?;
    let start = *at + SECTION_HEADER_LEN as usize;
    let end = start
        .checked_add(usize::try_from(len).map_err(|_| CapsuleDecodeError::LengthOverflow)?)
        .ok_or(CapsuleDecodeError::LengthOverflow)?;
    let body = input.get(start..end).ok_or(CapsuleDecodeError::Truncated {
        required: end,
        available: input.len(),
    })?;
    if tag != expected {
        return Err(if SECTION_TAGS.contains(&tag) {
            CapsuleDecodeError::SectionOutOfOrder { tag }
        } else {
            CapsuleDecodeError::UnknownSection { tag }
        });
    }
    *at = end;
    Ok(body)
}

/// Live values restored from one validated capsule.
#[derive(Debug, PartialEq, Eq)]
pub struct RestoredLifecycle {
    covered_log_sequence: u64,
    restored_fenced_sources: BTreeSet<ParticipantId>,
    conversation: ParticipantConversation,
    lifecycle: Option<DigestConversationState>,
    sealed: Option<SealedLifecycle>,
}

/// A sealed conversation's restored values: every retired identity and the
/// sequence-ledger high watermark its retired frontier owner reached.
#[derive(Debug, PartialEq, Eq)]
pub struct SealedLifecycle {
    retired: Vec<DigestRetiredIdentity>,
    ledger_high_watermark: u64,
}

impl SealedLifecycle {
    /// Every retired identity, strictly ascending by participant id.
    #[must_use]
    pub fn retired(&self) -> &[DigestRetiredIdentity] {
        &self.retired
    }

    /// The retired sequence ledger's high watermark at the seal.
    #[must_use]
    pub const fn ledger_high_watermark(&self) -> u64 {
        self.ledger_high_watermark
    }

    /// Takes every retired identity and the high watermark.
    #[must_use]
    pub fn into_parts(self) -> (Vec<DigestRetiredIdentity>, u64) {
        (self.retired, self.ledger_high_watermark)
    }
}

impl RestoredLifecycle {
    /// Restores one original observer source inside its associated joint capsule.
    ///
    /// Native producer fields and their exact projection are decoded together;
    /// neither the current frontier maximum nor a guessed scalar is a source.
    ///
    /// # Errors
    /// Refuses malformed, foreign or unassociated source evidence.
    #[doc(hidden)]
    pub fn restore_checkpoint_observer_projection(
        &self,
        bytes: &[u8],
        participant_id: ParticipantId,
    ) -> Result<super::ObserverProgressProjection, super::OwnerCheckpointRestoreError> {
        let (conversation_id, progress, source) =
            super::checkpoint_codec::decode_observer_source(bytes)?;
        if conversation_id != self.conversation.conversation_id()
            || source.participant_id() != participant_id
        {
            return Err(super::OwnerCheckpointRestoreError::Field(
                "observer_projection.capsule_association",
            ));
        }
        let high_watermark = match (&self.lifecycle, &self.sealed) {
            (Some(lifecycle), _) => lifecycle.frontiers().sequence().ledger().high_watermark(),
            (None, Some(sealed)) => sealed.ledger_high_watermark,
            (None, None) => {
                return Err(super::OwnerCheckpointRestoreError::Field(
                    "observer_projection.lifecycle",
                ));
            }
        };
        if u128::from(progress) > u128::from(high_watermark) + 1 {
            return Err(super::OwnerCheckpointRestoreError::Field(
                "observer_projection.delivery_coverage",
            ));
        }
        // The generation the source's binding epoch is bounded by. A live
        // conversation names every participant a source can belong to. A
        // sealed one names only the retired: the Died drain that sealed it
        // erased its last identity entirely, so a source of that participant
        // has no identity left to bound it and keeps the epoch the seal left.
        let generation = match (&self.lifecycle, &self.sealed) {
            (Some(lifecycle), _) => match lifecycle
                .participants()
                .iter()
                .find(|participant| participant.view().participant_id() == participant_id)
            {
                Some(super::RestoredParticipantLifecycle::Live { member, .. }) => {
                    Some(member.generation())
                }
                Some(super::RestoredParticipantLifecycle::Retired(retired)) => {
                    Some(retired.retired_generation())
                }
                None => {
                    return Err(super::OwnerCheckpointRestoreError::Field(
                        "observer_projection.participant",
                    ));
                }
            },
            (None, Some(sealed)) => sealed
                .retired
                .iter()
                .find(|retired| retired.participant_id() == participant_id)
                .map(super::RetiredIdentity::retired_generation),
            (None, None) => {
                return Err(super::OwnerCheckpointRestoreError::Field(
                    "observer_projection.lifecycle",
                ));
            }
        };
        if let Some(generation) = generation
            && source
                .binding_epoch()
                .is_some_and(|epoch| epoch.capability_generation.get() > generation.get())
        {
            return Err(super::OwnerCheckpointRestoreError::Field(
                "observer_projection.binding_generation",
            ));
        }
        Ok(super::ObserverProgressProjection::sourced(
            conversation_id,
            progress,
            source,
        ))
    }

    /// Restores one saved fenced predecessor through this joint lifecycle.
    ///
    /// The retained marker must be the exact delivered marker backing the
    /// participant's validated recovered origin. A source is issued once;
    /// refusals leave the lifecycle and issuance set unchanged.
    ///
    /// # Errors
    ///
    /// Refuses absent, foreign, duplicate or noncanonical source authority and
    /// propagates every native predecessor and successor validation failure.
    #[doc(hidden)]
    pub fn restore_checkpoint_fenced_attach(
        &mut self,
        source: super::FencedAttachCommitRestore,
        marker_record: &[u8],
    ) -> Result<super::FencedAttachCommit, super::OwnerCheckpointRestoreError> {
        use super::OwnerCheckpointRestoreError as Error;
        if self
            .restored_fenced_sources
            .contains(&source.participant_id)
        {
            return Err(Error::Field("fenced_source.duplicate"));
        }
        let lifecycle = self
            .lifecycle
            .as_ref()
            .ok_or(Error::Field("fenced_source.lifecycle"))?;
        let origin = lifecycle
            .participants()
            .iter()
            .find_map(|participant| match participant {
                super::RestoredParticipantLifecycle::Live {
                    member,
                    binding_origin: Some(origin),
                    ..
                } if member.participant_id() == source.participant_id => Some(*origin),
                _ => None,
            })
            .ok_or(Error::Field("fenced_source.binding_origin"))?;
        if origin.binding_epoch() != source.new_binding_epoch
            || origin.recovered_marker()
                != Some((source.marker_delivery_seq, source.prior_binding_epoch))
        {
            return Err(Error::Field("fenced_source.origin_association"));
        }
        let record = lifecycle
            .frontiers()
            .checkpoint_recovered_marker_record(origin)
            .ok_or(Error::Field("fenced_source.historical_marker"))?;
        if super::checkpoint_codec::encode_marker_source(&record) != marker_record {
            return Err(Error::Field("fenced_source.marker_bytes"));
        }
        let restored = source
            .restore_with_record(&record)
            .map_err(Error::Provenance);
        record.consume();
        let restored = restored?;
        self.restored_fenced_sources.insert(source.participant_id);
        Ok(restored)
    }

    /// The last log sequence whose effects the restored values include; a
    /// binding replays its log from the row after it.
    #[must_use]
    pub const fn covered_log_sequence(&self) -> u64 {
        self.covered_log_sequence
    }

    /// Moves the restored values out, for the binding to install.
    #[must_use]
    pub fn into_parts(self) -> RestoredLifecycleParts {
        RestoredLifecycleParts {
            conversation: self.conversation,
            lifecycle: self.lifecycle,
            sealed: self.sealed,
        }
    }

    /// The sealed conversation's retired identities and final watermark, when
    /// the capsule carried a seal.
    #[must_use]
    pub const fn sealed(&self) -> Option<&SealedLifecycle> {
        self.sealed.as_ref()
    }
}

/// The restored values, each for the binding to install where it keeps it.
/// It cannot be built outside the protocol: a binding receives one only from
/// [`RestoredLifecycle::into_parts`].
#[derive(Debug)]
#[non_exhaustive]
pub struct RestoredLifecycleParts {
    /// The conversation's event-sourced shell.
    pub conversation: ParticipantConversation,
    /// Every participant, both frontiers and the closure, validated together,
    /// with the fenced-marker issuance latch the capsule carried; `None` for a
    /// conversation captured before its first enrollment or after its seal.
    pub lifecycle: Option<DigestConversationState>,
    /// The retired identities and final watermark of a sealed conversation;
    /// `None` for one that is not sealed.
    pub sealed: Option<SealedLifecycle>,
}

/// Live values that cannot be captured into a capsule.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleCaptureError {
    /// The lent frontiers belong to another conversation than the shell.
    ForeignConversation,
    /// The lent participants are not strictly ascending by participant id.
    ParticipantOutOfOrder {
        /// The first participant id not above its predecessor.
        participant_id: ParticipantId,
    },
    /// The live conversation cannot be described back into restore input.
    Describe(ConversationDescribeError),
    /// A count or length does not fit its `u32` field.
    LengthOverflow,
    /// A view lent a lifecycle and a seal together; a sealed conversation has
    /// retired its frontier owner, so it has no lifecycle to lend.
    SealedBesideLifecycle,
    /// A sealed conversation lent a live participant.
    LiveParticipantSealed {
        /// The participant that is not retired.
        participant_id: ParticipantId,
    },
}

/// Stable canonical capsule-decode failure.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleDecodeError {
    /// Fewer bytes than the fixed envelope or selected field requires.
    Truncated {
        /// Minimum bytes required at the detected boundary.
        required: usize,
        /// Bytes supplied by the caller.
        available: usize,
    },
    /// The capsule does not begin with the assigned `LPCK` magic prefix.
    InvalidMagic,
    /// The capsule codec version is not supported by this crate.
    UnsupportedCodec {
        /// Presented codec major version.
        major: u16,
        /// Presented codec minor version.
        minor: u16,
    },
    /// The declared body length differs from the bytes supplied after the
    /// fixed header.
    NonCanonicalLength {
        /// Length declared in the envelope.
        declared_body_len: u32,
        /// Bytes supplied after the fixed header.
        actual_body_len: usize,
    },
    /// The body ended before a section the canonical order requires.
    MissingSection {
        /// Tag of the first absent section.
        tag: u16,
    },
    /// A known section appeared where the canonical order has another.
    SectionOutOfOrder {
        /// Tag of the misplaced section.
        tag: u16,
    },
    /// A section tag is unassigned.
    UnknownSection {
        /// Unrecognized section tag.
        tag: u16,
    },
    /// A section's declared length is not its canonical length.
    NonCanonicalSectionLength {
        /// Tag of the section.
        tag: u16,
        /// Length the section declared.
        declared: u32,
    },
    /// A complete section violates a field invariant.
    NonCanonicalSection {
        /// Tag of the section whose invariants failed.
        tag: u16,
    },
    /// The participants section is not strictly ascending by participant id.
    ParticipantOutOfOrder {
        /// The first participant id not above its predecessor.
        participant_id: ParticipantId,
    },
    /// A section names another conversation than the capsule header.
    ForeignConversation {
        /// Tag of the section naming the other conversation.
        tag: u16,
    },
    /// Bytes remain after the last canonical section.
    TrailingBytes {
        /// Bytes after the last section.
        trailing: usize,
    },
    /// A platform length conversion or offset addition overflowed.
    LengthOverflow,
}

/// A capsule whose sections do not form one valid lifecycle.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum CapsuleRestoreError {
    /// The shell's ordinal and genesis validation disagree.
    InconsistentConversation,
    /// The participants, frontiers and closure failed the joint
    /// participant-conversation validation.
    Lifecycle(ConversationStateRestoreError),
    /// A sealed conversation's retired identity failed its own restore.
    Sealed(StorageRestoreError),
    /// A sealed conversation carried a live participant.
    SealedLiveParticipant {
        /// The participant that is not retired.
        participant_id: ParticipantId,
    },
}

fn decode_conversation(body: &[u8]) -> Result<ConversationSection, CapsuleDecodeError> {
    if body.len() != CONVERSATION_SECTION_LEN as usize {
        return Err(CapsuleDecodeError::NonCanonicalSectionLength {
            tag: CONVERSATION_SECTION_TAG,
            declared: u32::try_from(body.len()).map_err(|_| CapsuleDecodeError::LengthOverflow)?,
        });
    }
    let next_event_ordinal = take_u64(body, 0)?;
    let flags = body.get(8).copied().ok_or(CapsuleDecodeError::Truncated {
        required: 9,
        available: body.len(),
    })?;
    if flags & !GENESIS_VALIDATED_FLAG != 0 {
        return Err(CapsuleDecodeError::NonCanonicalSection {
            tag: CONVERSATION_SECTION_TAG,
        });
    }
    let section = ConversationSection {
        next_event_ordinal,
        genesis_validated: flags & GENESIS_VALIDATED_FLAG != 0,
    };
    if !section.is_consistent() {
        return Err(CapsuleDecodeError::NonCanonicalSection {
            tag: CONVERSATION_SECTION_TAG,
        });
    }
    Ok(section)
}

fn take<const N: usize>(input: &[u8], start: usize) -> Result<[u8; N], CapsuleDecodeError> {
    let end = start
        .checked_add(N)
        .ok_or(CapsuleDecodeError::LengthOverflow)?;
    input
        .get(start..end)
        .ok_or(CapsuleDecodeError::Truncated {
            required: end,
            available: input.len(),
        })?
        .try_into()
        .map_err(|_| CapsuleDecodeError::LengthOverflow)
}

fn take_u16(input: &[u8], start: usize) -> Result<u16, CapsuleDecodeError> {
    take(input, start).map(u16::from_be_bytes)
}

fn take_u32(input: &[u8], start: usize) -> Result<u32, CapsuleDecodeError> {
    take(input, start).map(u32::from_be_bytes)
}

fn take_u64(input: &[u8], start: usize) -> Result<u64, CapsuleDecodeError> {
    take(input, start).map(u64::from_be_bytes)
}

/// An origin source that cannot join the validated capsule.
#[doc(hidden)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum BindingOriginSourceError {
    /// The canonical source reader refused the original bytes.
    Decode {
        /// Permanent participant selected by the source.
        participant_id: ParticipantId,
        /// Original canonical decode failure.
        error: CapsuleDecodeError,
    },
    /// Source identity, ordering or origin differs from the capsule.
    Association {
        /// Permanent participant at the refused association.
        participant_id: ParticipantId,
    },
    /// The complete capsule failed its existing joint validation.
    Restore(CapsuleRestoreError),
}

impl RestoredLifecycle {
    /// Restores saved cursor facts against this capsule's validated frontier and debt.
    ///
    /// # Errors
    ///
    /// Refuses absent lifecycle or debt and every mismatched saved episode fact.
    pub fn restore_cursor_episode(
        self,
        bytes: &[u8],
        observer_progress: u64,
    ) -> Result<RestoredLifecycleWithCursorEpisode, super::StorageRestoreError> {
        let lifecycle = self
            .lifecycle
            .as_ref()
            .ok_or(super::StorageRestoreError::CursorEpisode)?;
        let ClosureState::Owed { debt, .. } = lifecycle.closure() else {
            return Err(super::StorageRestoreError::CursorEpisode);
        };
        let episode = super::NonzeroDebtCursorEpisode::decode_checkpoint(
            bytes,
            lifecycle.frontiers(),
            debt,
            observer_progress,
        )?;
        Ok(RestoredLifecycleWithCursorEpisode {
            lifecycle: self,
            episode,
        })
    }
}

/// One capsule and the saved cursor facts validated against its own frontier.
#[derive(Debug)]
pub struct RestoredLifecycleWithCursorEpisode {
    lifecycle: RestoredLifecycle,
    episode: super::NonzeroDebtCursorEpisode,
}
impl RestoredLifecycleWithCursorEpisode {
    /// Moves both jointly validated values to their operational owners.
    #[must_use]
    pub fn into_parts(self) -> (RestoredLifecycleParts, super::NonzeroDebtCursorEpisode) {
        (self.lifecycle.into_parts(), self.episode)
    }
}

impl LifecycleCapsule {
    /// Captures borrowed participants using their retained canonical origin sources.
    ///
    /// # Errors
    ///
    /// Refuses missing, malformed, unordered or foreign sources, then propagates
    /// the ordinary capsule-capture error. Decoded origins must match the sealed
    /// frontier and the exact member and binding before any view borrows them.
    pub fn capture_with_binding_sources(
        view: &LifecycleView<'_>,
        sources: &[(ParticipantId, &[u8])],
    ) -> Result<Self, BindingOriginCaptureError> {
        let Some(lifecycle) = view.lifecycle.as_ref() else {
            if let Some(&(participant_id, _)) = sources.first() {
                return Err(BindingOriginCaptureError::Association { participant_id });
            }
            return Self::capture(view).map_err(BindingOriginCaptureError::Capture);
        };
        let mut remaining = sources.iter().peekable();
        let mut origins = Vec::new();
        for participant in &lifecycle.participants {
            match participant {
                DigestParticipantView::Retired(_) => origins.push(None),
                DigestParticipantView::Live {
                    member,
                    binding,
                    binding_origin,
                    ..
                } => {
                    let participant_id = member.participant_id();
                    let origin_required = !matches!(binding, super::BindingState::Detached)
                        || member.latest_terminal().is_some()
                        || binding_origin.is_some();
                    let Some(&&(source_id, bytes)) = remaining.peek() else {
                        if origin_required {
                            return Err(BindingOriginCaptureError::Association { participant_id });
                        }
                        origins.push(None);
                        continue;
                    };
                    if source_id != participant_id {
                        if origin_required || source_id < participant_id {
                            return Err(BindingOriginCaptureError::Association { participant_id });
                        }
                        origins.push(None);
                        continue;
                    }
                    remaining.next();
                    let origin = super::checkpoint_codec::decode_binding_origin_source(bytes)
                        .map_err(|error| BindingOriginCaptureError::Decode {
                            participant_id,
                            error,
                        })?;
                    super::storage::validate_binding_origin(origin, member, **binding).map_err(
                        |error| BindingOriginCaptureError::Validation {
                            participant_id,
                            error,
                        },
                    )?;
                    if binding_origin.is_some_and(|expected| *expected != origin) {
                        return Err(BindingOriginCaptureError::Association { participant_id });
                    }
                    origins.push(Some(origin));
                }
            }
        }
        if let Some(&(participant_id, _)) = remaining.next() {
            return Err(BindingOriginCaptureError::Association { participant_id });
        }
        let participants = lifecycle
            .participants
            .iter()
            .zip(&origins)
            .map(|(participant, origin)| match participant {
                DigestParticipantView::Live {
                    member,
                    binding,
                    detach_cell,
                    ..
                } => DigestParticipantView::Live {
                    member,
                    binding,
                    detach_cell,
                    binding_origin: origin.as_ref(),
                },
                DigestParticipantView::Retired(retired) => DigestParticipantView::Retired(retired),
            })
            .collect();
        let capsule = Self::capture(&LifecycleView {
            sealed: None,
            conversation: view.conversation,
            covered_log_sequence: view.covered_log_sequence,
            lifecycle: Some(ConversationLifecycleView {
                participants,
                frontiers: lifecycle.frontiers,
                retained_record_limit: lifecycle.retained_record_limit,
                closure: lifecycle.closure,
            }),
        })
        .map_err(BindingOriginCaptureError::Capture)?;
        let canonical = capsule.encode_canonical();
        capsule
            .restore()
            .map_err(BindingOriginCaptureError::Restore)?;
        Self::decode_canonical(&canonical).map_err(BindingOriginCaptureError::Canonical)
    }
}

/// Canonical source decoding and exact live capture association failures.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum BindingOriginCaptureError {
    /// The complete capsule failed its native joint history and frontier validation.
    Restore(CapsuleRestoreError),
    /// The canonical capsule reader refused the captured bytes.
    Canonical(CapsuleDecodeError),
    /// The original canonical source reader refused these bytes.
    Decode {
        /// The permanent participant selected by this source.
        participant_id: ParticipantId,
        /// The original canonical source decode failure.
        error: CapsuleDecodeError,
    },
    /// The source does not belong to this participant and sealed frontier.
    Association {
        /// The permanent participant at the refused association.
        participant_id: ParticipantId,
    },
    /// The source does not match the existing member and binding.
    Validation {
        /// The permanent participant selected by this source.
        participant_id: ParticipantId,
        /// The original native member and binding validation failure.
        error: super::StorageRestoreError,
    },
    /// The ordinary capsule writer refused the borrowed values.
    Capture(CapsuleCaptureError),
}