liminal-protocol 0.11.0

//! Pins for the lifecycle capsule's participants, frontiers and closure
//! sections: a live conversation describes itself and restores to itself, the
//! capsule round-trips it with and without the fenced-marker latch, a restored
//! latch refuses a second fenced take exactly as the live one does, a carried
//! latch no live take set is refused at restore, and every new refusal is
//! named.

use alloc::{vec, vec::Vec};

use proptest::prelude::*;

use crate::algebra::WideResourceVector;
use crate::wire::{AttachSecret, BindingEpoch, CloseCause};

use super::checkpoint_codec::{DigestDescription, decode_description, encode_description};
use super::claim_frontier_tests::detached_dcr_fixture;
use super::conversation_tests::{CONVERSATION_ID as SHELL_CONVERSATION_ID, build_operation};
use super::storage::{ParticipantConversationDescription, describe_conversation};
use super::storage_tests::{
    PARTICIPANT_ID, binding, died_terminal, enrollment_bound_snapshot, epoch,
    ordinary_bound_conversation, ordinary_detached_conversation, owed,
    recovered_bound_with_active_quartet, recovered_conversation, replayed_enrollment_commit,
};
use super::{
    ActiveBinding, AttachedLifecycleRecord, AttachedRecordPosition, BindingFateTerminalRestore,
    BindingOrigin, BindingState, BindingStateRestore, CapsuleCaptureError, CapsuleDecodeError,
    CapsuleRestoreError, ClaimFrontiers, ClosureState, ClosureStateRestore,
    CommittedBindingTerminal, CommittedBindingTerminalRestore, ConversationDecision,
    ConversationDescribeError, ConversationGenesis, ConversationLifecycleView,
    ConversationStateRestoreError, DetachCellRestore, DetachedCredentialRecovery,
    DetachedCredentialRecoveryRestore, DetachedCursorReleaseProvenanceRestore,
    EnrollmentFingerprint, FrontierBinding, LifecycleCapsule, LifecycleView, LiveIdentityRestore,
    MarkerCursorProgressRestore, MarkerDeliveryRestore, OrdinaryBindingAuthorityRestore,
    OrdinaryBindingFateRestore, ParticipantConversation, ParticipantConversationRestore,
    ParticipantLifecycleRestore, RestoredParticipantLifecycle, StorageRestoreError, StoredEdge,
    StoredEdgeRestore,
};

type Digest = [u8; 32];
type DigestRestore<const N: usize> =
    ParticipantConversationRestore<[u8; N], [u8; N], [u8; N], [u8; N]>;
type DigestParticipant<const N: usize> =
    RestoredParticipantLifecycle<[u8; N], [u8; N], [u8; N], [u8; N]>;

/// The detached participant of the recovery fixture.
const PARTICIPANT: u64 = 0;
/// The recovery fixture's delivered marker.
const MARKER: u64 = 1;

// ---------------------------------------------------------------------------
// Fixtures
// ---------------------------------------------------------------------------

/// A whole detached-credential-recovery conversation: participant 0 died on
/// its one epoch after a marker was delivered to it, and the closure stands
/// on the recovery that delivery and that death selected. Built for any
/// conversation, so it can sit under any shell.
fn dcr_restore(conversation_id: u64) -> Result<DigestRestore<32>, &'static str> {
    let (mut frontiers, sequence_ledger, order_ledger, _, binding_epoch) = detached_dcr_fixture()?;
    frontiers.conversation_id = conversation_id;
    for fact in &mut frontiers.historical_marker_deliveries {
        fact.conversation_id = conversation_id;
    }
    let ended = ActiveBinding {
        participant_id: PARTICIPANT,
        conversation_id,
        binding_epoch,
    };
    let terminal = CommittedBindingTerminalRestore {
        binding: ended,
        cause: CloseCause::ConnectionLost,
        transaction_order: 1,
        delivery_seq: 2,
    };
    let delivery = MarkerDeliveryRestore {
        participant_id: PARTICIPANT,
        binding_epoch,
        marker_delivery_seq: MARKER,
    };
    Ok(ParticipantConversationRestore {
        participants: vec![ParticipantLifecycleRestore::Live {
            identity: LiveIdentityRestore {
                participant_id: PARTICIPANT,
                conversation_id,
                generation: binding_epoch.capability_generation,
                attach_secret: AttachSecret::new([0xA1; 32]),
                cursor: 0,
                enrollment_fingerprint: EnrollmentFingerprint::new([0xE1; 32]),
                latest_terminal: Some(terminal),
            },
            binding: BindingStateRestore::Detached,
            binding_origin: Some(BindingOrigin::unfenced(
                AttachedLifecycleRecord::from_binding(ended, AttachedRecordPosition::new(0, 0)),
            )),
            detach_cell: DetachCellRestore::Empty,
        }],
        frontiers,
        sequence_ledger,
        order_ledger,
        closure: ClosureStateRestore::Owed {
            debt: WideResourceVector::new(1, 1),
            edge: StoredEdgeRestore::DetachedCredentialRecovery(
                DetachedCredentialRecoveryRestore {
                    participant_id: PARTICIPANT,
                    marker_delivery_seq: MARKER,
                    prior_binding_epoch: binding_epoch,
                    resulting_floor: 1,
                    terminal: BindingFateTerminalRestore::Committed(terminal),
                    progress: MarkerCursorProgressRestore {
                        conversation_id,
                        participant_id: PARTICIPANT,
                        binding_epoch,
                        through_seq: MARKER,
                        marker_delivery_seq: MARKER,
                        delivery,
                    },
                },
            ),
        },
    })
}

/// The live values a binding would hold for the recovery conversation.
struct LiveDcr {
    participants: Vec<DigestParticipant<32>>,
    frontiers: ClaimFrontiers,
    closure: ClosureState,
    recovery: DetachedCredentialRecovery,
    retained_record_limit: u64,
}

/// Restores the recovery conversation and, when `latched`, takes its one
/// fenced marker as a fenced attach mint would, setting the issuance latch.
fn live_dcr(conversation_id: u64, latched: bool) -> Result<LiveDcr, &'static str> {
    let restore = dcr_restore(conversation_id)?;
    let retained_record_limit = restore.frontiers.retained_record_limit;
    let (participants, mut frontiers, closure) = restore
        .restore()
        .map_err(|_| "the joint recovery conversation restores")?
        .into_parts();
    let ClosureState::Owed {
        edge: StoredEdge::DetachedCredentialRecovery(recovery),
        ..
    } = closure
    else {
        return Err("the recovery conversation stands on its recovery");
    };
    if latched {
        frontiers
            .take_fenced_marker_record(recovery)
            .ok_or("the recovery takes its one fenced marker")?
            .consume();
    }
    Ok(LiveDcr {
        participants,
        frontiers,
        closure,
        recovery,
        retained_record_limit,
    })
}

/// A shell whose genesis validation committed.
fn validated_shell(conversation_id: u64) -> Result<ParticipantConversation, &'static str> {
    let fresh = ParticipantConversation::from_genesis(ConversationGenesis::new(conversation_id));
    let ConversationDecision::Commit(commit) = fresh.decide_genesis_validation() else {
        return Err("a fresh genesis selects its one-shot event");
    };
    Ok(commit.commit())
}

/// The shell after its genesis and the generated operations, through the live
/// path exactly as `any_prefix_replay_produces_a_valid_shell_state` builds it.
fn shell_after(operations: &[(u8, [u64; 5])]) -> Result<ParticipantConversation, &'static str> {
    let mut shell = validated_shell(SHELL_CONVERSATION_ID)?;
    for &(kind, values) in operations {
        let ConversationDecision::Commit(commit) =
            shell.decide_operation(build_operation(kind, values))
        else {
            return Err("a validated shell accepts every same-conversation operation");
        };
        shell = commit.commit();
    }
    Ok(shell)
}

/// The joint state a capsule restores.
type RestoredState = super::ParticipantConversationState<Digest, Digest, Digest, Digest>;

fn lifecycle_view(live: &LiveDcr) -> ConversationLifecycleView<'_> {
    ConversationLifecycleView {
        participants: live
            .participants
            .iter()
            .map(RestoredParticipantLifecycle::view)
            .collect(),
        frontiers: &live.frontiers,
        retained_record_limit: live.retained_record_limit,
        closure: &live.closure,
    }
}

fn capture(
    shell: &ParticipantConversation,
    live: &LiveDcr,
    covered: u64,
) -> Result<LifecycleCapsule, CapsuleCaptureError> {
    LifecycleCapsule::capture(&LifecycleView {
        conversation: shell,
        lifecycle: Some(lifecycle_view(live)),
        covered_log_sequence: covered,
    })
}

/// Describes the live recovery conversation, latch included.
fn describe_live(live: &LiveDcr) -> Result<DigestDescription<32>, &'static str> {
    let view = lifecycle_view(live);
    describe_conversation(
        &view.participants,
        view.frontiers,
        view.retained_record_limit,
        live.closure,
    )
    .map_err(|_| "the live recovery conversation describes itself")
}

/// The latch a fenced take of the recovery fixture's marker sets.
const fn expected_latch(live: &LiveDcr, latched: bool) -> Option<(u64, u64, BindingEpoch)> {
    if latched {
        Some((PARTICIPANT, MARKER, live.recovery.prior_binding_epoch()))
    } else {
        None
    }
}

/// Capture, encode, decode, restore; asserts the decoded capsule and its
/// bytes are the captured ones, and answers the restored shell and state.
fn capsule_round_trip(
    shell: &ParticipantConversation,
    live: &LiveDcr,
    covered: u64,
) -> Result<(ParticipantConversation, RestoredState), &'static str> {
    let captured = capture(shell, live, covered).map_err(|_| "the live lifecycle captures")?;
    let bytes = captured.encode_canonical();
    let decoded = LifecycleCapsule::decode_canonical(&bytes).map_err(|_| "the capsule decodes")?;
    assert_eq!(
        decoded, captured,
        "decode(encode(capture(x))) == capture(x)"
    );
    assert_eq!(decoded.encode_canonical(), bytes, "the bytes are canonical");
    let restored = decoded.restore().map_err(|_| "the capsule restores")?;
    assert_eq!(restored.covered_log_sequence(), covered);
    let parts = restored.into_parts();
    let lifecycle = parts.lifecycle.ok_or("the lifecycle comes back")?;
    Ok((parts.conversation, lifecycle))
}

/// Splits framed sections into `(tag, body)` pairs.
fn split_sections(mut bytes: &[u8]) -> Result<Vec<(u16, Vec<u8>)>, &'static str> {
    let mut sections = Vec::new();
    while !bytes.is_empty() {
        let tag = u16::from_be_bytes(
            bytes
                .get(0..2)
                .ok_or("tag")?
                .try_into()
                .map_err(|_| "tag width")?,
        );
        let len = u32::from_be_bytes(
            bytes
                .get(2..6)
                .ok_or("length")?
                .try_into()
                .map_err(|_| "length width")?,
        );
        let end = 6 + usize::try_from(len).map_err(|_| "length fits")?;
        sections.push((tag, bytes.get(6..end).ok_or("body")?.to_vec()));
        bytes = bytes.get(end..).ok_or("rest")?;
    }
    Ok(sections)
}

fn frame(tag: u16, body: &[u8]) -> Result<Vec<u8>, &'static str> {
    let mut framed = tag.to_be_bytes().to_vec();
    framed.extend_from_slice(
        &u32::try_from(body.len())
            .map_err(|_| "body fits")?
            .to_be_bytes(),
    );
    framed.extend_from_slice(body);
    Ok(framed)
}

/// A v1.0 capsule for a validated shell at ordinal 1, followed by `sections`.
fn envelope(conversation_id: u64, sections: &[u8]) -> Result<Vec<u8>, &'static str> {
    let mut body = frame(1, &[0, 0, 0, 0, 0, 0, 0, 1, 1])?;
    body.extend_from_slice(sections);
    let mut bytes = b"LPCK".to_vec();
    bytes.extend_from_slice(&1_u16.to_be_bytes());
    bytes.extend_from_slice(&0_u16.to_be_bytes());
    bytes.extend_from_slice(&conversation_id.to_be_bytes());
    bytes.extend_from_slice(&0_u64.to_be_bytes());
    bytes.extend_from_slice(
        &u32::try_from(body.len())
            .map_err(|_| "body fits")?
            .to_be_bytes(),
    );
    bytes.extend(body);
    Ok(bytes)
}

/// The three framed sections of a description, as `(tag, body)` pairs.
fn sections_of(description: &DigestDescription<32>) -> Result<Vec<(u16, Vec<u8>)>, &'static str> {
    split_sections(&encode_description(description).ok_or("the description encodes")?)
}

fn join(sections: &[(u16, Vec<u8>)]) -> Result<Vec<u8>, &'static str> {
    let mut joined = Vec::new();
    for (tag, body) in sections {
        joined.extend(frame(*tag, body)?);
    }
    Ok(joined)
}

fn decode(bytes: &[u8]) -> Option<CapsuleDecodeError> {
    LifecycleCapsule::decode_canonical(bytes).err()
}

/// Asserts one described participant equals the live getters it came from.
fn assert_participant_describes_live<const N: usize>(
    described: &ParticipantLifecycleRestore<[u8; N], [u8; N], [u8; N], [u8; N]>,
    live: &DigestParticipant<N>,
) -> Result<(), &'static str> {
    match (described, live) {
        (
            ParticipantLifecycleRestore::Live {
                identity,
                binding,
                binding_origin,
                detach_cell,
            },
            RestoredParticipantLifecycle::Live {
                member,
                binding: live_binding,
                binding_origin: live_origin,
                detach_cell: live_cell,
            },
        ) => {
            assert_eq!(identity.participant_id, member.participant_id());
            assert_eq!(identity.conversation_id, member.conversation_id());
            assert_eq!(identity.generation, member.generation());
            assert_eq!(identity.attach_secret, member.attach_secret());
            assert_eq!(identity.cursor, member.cursor());
            assert_eq!(
                &identity.enrollment_fingerprint,
                member.enrollment_fingerprint()
            );
            assert_eq!(
                identity.latest_terminal,
                member
                    .latest_terminal()
                    .map(CommittedBindingTerminal::storage_description)
            );
            match (binding, live_binding) {
                (BindingStateRestore::Detached, BindingState::Detached) => {}
                (BindingStateRestore::Bound(described), BindingState::Bound(live)) => {
                    assert_eq!(described, live);
                }
                (
                    BindingStateRestore::PendingFinalization(described),
                    BindingState::PendingFinalization(live),
                ) => assert_eq!(*described, live.storage_description()),
                _ => return Err("the described binding slot is another variant"),
            }
            assert_eq!(binding_origin.as_ref(), live_origin.as_ref());
            assert_eq!(detach_cell, &live_cell.storage_description());
        }
        (
            ParticipantLifecycleRestore::Retired(described),
            RestoredParticipantLifecycle::Retired(live),
        ) => {
            assert_eq!(described.participant_id, live.participant_id());
            assert_eq!(described.conversation_id, live.conversation_id());
            assert_eq!(described.retired_generation, live.retired_generation());
            assert_eq!(
                &described.enrollment_fingerprint,
                live.enrollment_fingerprint()
            );
            assert_eq!(described.leave_attempt_token, live.leave_attempt_token());
            assert_eq!(
                &described.leave_request_verifier,
                live.leave_request_verifier()
            );
            assert_eq!(&described.leave_fingerprint, live.leave_fingerprint());
            assert_eq!(
                described.left_transaction_order,
                live.left_admission_order().transaction_order()
            );
            assert_eq!(
                described.committed_result.left_delivery_seq,
                live.committed_result().left_delivery_seq()
            );
        }
        _ => return Err("the described participant is another variant"),
    }
    Ok(())
}

/// Asserts every described field equals the live getter it was taken from.
fn assert_describes_live<const N: usize>(
    description: &ParticipantConversationDescription<[u8; N], [u8; N], [u8; N], [u8; N]>,
    participants: &[DigestParticipant<N>],
    frontiers: &ClaimFrontiers,
    closure: ClosureState,
) -> Result<(), &'static str> {
    let restore = &description.restore;
    assert_eq!(restore.participants.len(), participants.len());
    for (described, live) in restore.participants.iter().zip(participants) {
        assert_participant_describes_live(described, live)?;
    }
    let described = &restore.frontiers;
    assert_eq!(described.conversation_id, frontiers.conversation_id());
    assert_eq!(
        described.active_identities.as_slice(),
        frontiers.active_identities().participants()
    );
    assert_eq!(
        described.identity_slot_limit,
        frontiers.identity_slot_limit()
    );
    assert_eq!(described.retained_floor, frontiers.retained_floor());
    assert_eq!(
        described.retained_records.as_slice(),
        frontiers.retained_records()
    );
    assert_eq!(restore.sequence_ledger, frontiers.sequence().ledger());
    assert_eq!(restore.order_ledger, frontiers.order().ledger());
    match (restore.closure, closure) {
        (ClosureStateRestore::Clear, ClosureState::Clear) => Ok(()),
        (ClosureStateRestore::Owed { debt, .. }, ClosureState::Owed { debt: live, .. }) => {
            assert_eq!(debt, live.value());
            Ok(())
        }
        _ => Err("the described closure is another variant"),
    }
}

// ---------------------------------------------------------------------------
// Pin (a): live -> describe -> restore == live
// ---------------------------------------------------------------------------

/// Every whole-conversation fixture `storage_tests` proves restorable, plus
/// the recovery conversation, in the digest width each was written for.
fn storage_fixtures() -> Vec<DigestRestore<4>> {
    let bound_epoch = epoch(2, 8);
    let recovered_epoch = epoch(3, 9);
    let continuous = StoredEdgeRestore::ParticipantCursorProgressContinuous {
        participant_id: PARTICIPANT_ID,
        binding_epoch: bound_epoch,
        through_seq: 12,
        authority: OrdinaryBindingAuthorityRestore {
            binding: binding(PARTICIPANT_ID, bound_epoch),
            through_seq: 12,
        },
    };
    let projection = ClosureStateRestore::Owed {
        debt: WideResourceVector::new(2, 16),
        edge: StoredEdgeRestore::ObserverProjection { through_seq: 14 },
    };
    let compaction = ClosureStateRestore::Owed {
        debt: WideResourceVector::new(2, 16),
        edge: StoredEdgeRestore::PhysicalCompaction {
            from_floor: 12,
            through_seq: 14,
        },
    };
    vec![
        ordinary_bound_conversation(owed(continuous)),
        enrollment_bound_snapshot(&replayed_enrollment_commit(AttachedRecordPosition::new(
            0, 1,
        ))),
        recovered_conversation(
            FrontierBinding::Bound(recovered_epoch),
            Some(bound_epoch),
            projection,
        ),
        recovered_bound_with_active_quartet(Some(bound_epoch), projection),
        recovered_bound_with_active_quartet(Some(bound_epoch), compaction),
    ]
}

#[test]
fn every_restorable_storage_fixture_describes_and_restores_to_itself() -> Result<(), &'static str> {
    for fixture in storage_fixtures() {
        let retained_record_limit = fixture.frontiers.retained_record_limit;
        let live = fixture
            .restore()
            .map_err(|_| "every storage fixture restores")?;
        let description = live
            .describe(retained_record_limit)
            .map_err(|_| "every storage fixture describes itself")?;
        assert_describes_live(
            &description,
            live.participants(),
            live.frontiers(),
            live.closure(),
        )?;
        assert_eq!(description.fenced_marker_issued, None);

        let sections =
            split_sections(&encode_description(&description).ok_or("the description encodes")?)?;
        let [(2, participants), (3, frontiers), (4, closure)] = sections.as_slice() else {
            return Err("three sections in their fixed order");
        };
        let decoded = decode_description::<4>(participants, frontiers, closure)
            .map_err(|_| "the encoded description decodes")?;
        assert_eq!(
            decoded, description,
            "the codec round-trips four-byte digests"
        );

        let restored = decoded
            .into_state()
            .map_err(|_| "the described fixture restores")?;
        assert_eq!(restored.participants(), live.participants());
        assert_eq!(restored.frontiers(), live.frontiers());
        assert_eq!(restored.closure(), live.closure());
        assert_eq!(restored, live);
    }
    Ok(())
}

#[test]
fn the_recovery_conversation_describes_its_exact_restore_input() -> Result<(), &'static str> {
    let input = dcr_restore(SHELL_CONVERSATION_ID)?;
    let retained_record_limit = input.frontiers.retained_record_limit;
    let live = input
        .clone()
        .restore()
        .map_err(|_| "the recovery conversation restores")?;
    let description = live
        .describe(retained_record_limit)
        .map_err(|_| "the recovery conversation describes itself")?;
    assert_describes_live(
        &description,
        live.participants(),
        live.frontiers(),
        live.closure(),
    )?;
    assert_eq!(description.restore, input, "the input comes back unchanged");
    assert_eq!(description.fenced_marker_issued, None);
    let restored = description
        .into_state()
        .map_err(|_| "the description restores")?;
    assert_eq!(restored, live);
    Ok(())
}

#[test]
fn a_latched_live_conversation_describes_and_restores_its_latch() -> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, true)?;
    let description = describe_live(&live)?;
    assert_describes_live(
        &description,
        &live.participants,
        &live.frontiers,
        live.closure,
    )?;
    assert_eq!(
        description.fenced_marker_issued,
        expected_latch(&live, true)
    );
    let restored = description
        .clone()
        .into_state()
        .map_err(|_| "the latched description restores")?;
    assert_eq!(restored.participants(), live.participants.as_slice());
    assert_eq!(
        restored.frontiers(),
        &live.frontiers,
        "the latch is installed"
    );
    assert_eq!(restored.closure(), live.closure);

    let public = description
        .restore
        .restore()
        .map_err(|_| "the public route restores the same input")?;
    assert_ne!(
        public.frontiers(),
        &live.frontiers,
        "the public route keeps its forced None"
    );
    Ok(())
}

#[test]
fn a_detached_cursor_release_cannot_be_described() -> Result<(), &'static str> {
    let bound_epoch = epoch(2, 8);
    let fixture = ordinary_detached_conversation(owed(StoredEdgeRestore::DetachedCursorRelease {
        participant_id: PARTICIPANT_ID,
        last_dead_binding_epoch: bound_epoch,
        provenance: DetachedCursorReleaseProvenanceRestore::Ordinary(OrdinaryBindingFateRestore {
            authority: OrdinaryBindingAuthorityRestore {
                binding: binding(PARTICIPANT_ID, bound_epoch),
                through_seq: 12,
            },
            terminal: died_terminal(PARTICIPANT_ID, bound_epoch, 9, 13),
            resulting_floor: 14,
        }),
    }));
    let retained_record_limit = fixture.frontiers.retained_record_limit;
    let live = fixture
        .restore()
        .map_err(|_| "the ordinary cursor release restores")?;
    assert_eq!(
        live.describe(retained_record_limit).err(),
        Some(ConversationDescribeError::DetachedCursorReleaseProvenance)
    );
    Ok(())
}

#[test]
fn a_recovery_without_its_member_cannot_be_described() -> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, false)?;
    let views: Vec<super::ParticipantLifecycleView<'_, Digest, Digest, Digest, Digest>> =
        Vec::new();
    assert_eq!(
        describe_conversation(
            &views,
            &live.frontiers,
            live.retained_record_limit,
            live.closure
        )
        .err(),
        Some(ConversationDescribeError::EdgeTerminal)
    );
    Ok(())
}

// ---------------------------------------------------------------------------
// Pin (b): decode(encode(capture(x))) == capture(x), restore == live
// ---------------------------------------------------------------------------

#[test]
fn the_recovery_lifecycle_round_trips_with_and_without_the_latch() -> Result<(), &'static str> {
    let shell = validated_shell(SHELL_CONVERSATION_ID)?;
    for latched in [false, true] {
        let live = live_dcr(SHELL_CONVERSATION_ID, latched)?;
        let (conversation, lifecycle) = capsule_round_trip(&shell, &live, 7)?;
        assert_eq!(conversation, shell);
        assert_eq!(lifecycle.participants(), live.participants.as_slice());
        assert_eq!(lifecycle.frontiers(), &live.frontiers);
        assert_eq!(lifecycle.closure(), live.closure);
        assert_eq!(
            lifecycle
                .frontiers()
                .describe(live.retained_record_limit)
                .fenced_marker_issued,
            expected_latch(&live, latched)
        );
    }
    Ok(())
}

proptest! {
    #[test]
    fn any_prefix_shell_with_the_recovery_lifecycle_round_trips_through_the_capsule(
        raw_operations in proptest::collection::vec(
            (0_u8..6, [any::<u64>(), any::<u64>(), any::<u64>(), any::<u64>(), any::<u64>()]),
            0..12,
        ),
        latched in any::<bool>(),
        covered in any::<u64>(),
    ) {
        let shell = shell_after(&raw_operations).map_err(TestCaseError::fail)?;
        let live = live_dcr(SHELL_CONVERSATION_ID, latched).map_err(TestCaseError::fail)?;
        let (conversation, lifecycle) =
            capsule_round_trip(&shell, &live, covered).map_err(TestCaseError::fail)?;
        prop_assert_eq!(&conversation, &shell);
        prop_assert_eq!(lifecycle.participants(), live.participants.as_slice());
        prop_assert_eq!(lifecycle.frontiers(), &live.frontiers);
        prop_assert_eq!(lifecycle.closure(), live.closure);
    }
}

// ---------------------------------------------------------------------------
// Pin (c): a restored latch refuses a second fenced take exactly as live
// ---------------------------------------------------------------------------

#[test]
fn a_carried_fenced_latch_refuses_a_second_fenced_take_after_restore() -> Result<(), &'static str> {
    let shell = validated_shell(SHELL_CONVERSATION_ID)?;
    let mut live = live_dcr(SHELL_CONVERSATION_ID, true)?;
    let (_, lifecycle) = capsule_round_trip(&shell, &live, 3)?;
    let (_, mut restored, _) = lifecycle.into_parts();
    let recovery = live.recovery;

    assert!(
        live.frontiers.take_fenced_marker_record(recovery).is_none(),
        "live refuses a second fenced take"
    );
    assert!(
        restored.take_fenced_marker_record(recovery).is_none(),
        "the restored latch refuses it the same way"
    );
    assert_eq!(
        restored, live.frontiers,
        "both refusals leave the same state"
    );

    // The same described input through the public route has no latch, so it
    // would grant the take: the carry is what refuses it.
    let mut public = describe_live(&live)?
        .restore
        .restore()
        .map_err(|_| "the public route restores")?
        .into_parts()
        .1;
    public
        .take_fenced_marker_record(recovery)
        .ok_or("without the carried latch the take is granted")?
        .consume();
    Ok(())
}

// ---------------------------------------------------------------------------
// Pin (c'): a carried latch is checked against the restored state
// ---------------------------------------------------------------------------

/// The latched recovery capsule's bytes with only the latch edited to `latch`.
///
/// The latch is the frontiers section's last field. Its option flag sits at the
/// offset where the unlatched encoding of the same state ends its frontiers
/// body with the absent flag, so the latched body is cut there and the
/// replacement latch's own encoded bytes are spliced in. Everything before the
/// flag is the real latched capsule's, byte for byte.
fn latched_bytes_with_latch(
    latch: Option<(u64, u64, BindingEpoch)>,
) -> Result<Vec<u8>, &'static str> {
    let described = describe_live(&live_dcr(SHELL_CONVERSATION_ID, true)?)?;
    let sections = sections_of(&described)?;
    let [participants, (3, frontiers), closure] = sections.as_slice() else {
        return Err("three sections with the frontiers second");
    };
    let mut unlatched = described.clone();
    unlatched.fenced_marker_issued = None;
    let flag = sections_of(&unlatched)?
        .get(1)
        .map(|(_, body)| body.len())
        .ok_or("the unlatched frontiers section")?
        .checked_sub(1)
        .ok_or("the unlatched frontiers body ends in its absent flag")?;
    assert_eq!(frontiers.get(flag), Some(&1), "the real latch is present");

    let mut replaced = described;
    replaced.fenced_marker_issued = latch;
    let replacement = sections_of(&replaced)?;
    let (_, replacement_body) = replacement.get(1).ok_or("the replacement frontiers")?;
    let mut body = frontiers
        .get(..flag)
        .ok_or("the frontiers before the latch")?
        .to_vec();
    assert_eq!(
        replacement_body.get(..flag),
        Some(body.as_slice()),
        "only the latch bytes are edited"
    );
    body.extend_from_slice(
        replacement_body
            .get(flag..)
            .ok_or("the replacement latch bytes")?,
    );
    envelope(
        SHELL_CONVERSATION_ID,
        &join(&[participants.clone(), (3, body), closure.clone()])?,
    )
}

fn restore_bytes(bytes: &[u8]) -> Result<Option<CapsuleRestoreError>, &'static str> {
    let capsule = LifecycleCapsule::decode_canonical(bytes)
        .map_err(|_| "a latch edit is not a byte fault")?;
    Ok(capsule.restore().err())
}

const LATCH_REFUSED: CapsuleRestoreError = CapsuleRestoreError::Lifecycle(
    ConversationStateRestoreError::Storage(StorageRestoreError::FencedMarkerLatch),
);

#[test]
fn a_foreign_fenced_latch_is_refused_by_name_at_restore() -> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, true)?;
    let genuine = expected_latch(&live, true).ok_or("the fixture is latched")?;
    let (participant, marker, prior_epoch) = genuine;

    // The genuine latch, spliced back through the same edit, restores.
    assert_eq!(
        restore_bytes(&latched_bytes_with_latch(Some(genuine))?)?,
        None
    );

    let mut foreign_epoch = prior_epoch;
    foreign_epoch.connection_incarnation.connection_ordinal ^= 1;
    assert_ne!(foreign_epoch, prior_epoch);
    for foreign in [
        (participant + 1, marker, prior_epoch),
        (participant, marker + 1, prior_epoch),
        (participant, marker, foreign_epoch),
    ] {
        assert_eq!(
            restore_bytes(&latched_bytes_with_latch(Some(foreign))?)?,
            Some(LATCH_REFUSED),
            "a latch no live take set is refused, not installed to refuse the real take"
        );
    }
    Ok(())
}

#[test]
fn a_fenced_latch_under_a_non_recovery_edge_is_refused_by_name() -> Result<(), &'static str> {
    let fixture = storage_fixtures()
        .into_iter()
        .next()
        .ok_or("the ordinary bound fixture")?;
    let description = ParticipantConversationDescription {
        restore: fixture,
        fenced_marker_issued: Some((PARTICIPANT_ID, 12, epoch(2, 8))),
    };
    assert_eq!(
        description.into_state().err(),
        Some(ConversationStateRestoreError::Storage(
            StorageRestoreError::FencedMarkerLatch
        ))
    );
    Ok(())
}

/// The residue: a latch flipped to `None` cannot be refused. A fenced take
/// writes nothing but the latch, so the flipped capsule is byte for byte the
/// capsule of the same state before any take, which is a legitimate capsule
/// and restores granting the take.
#[test]
fn a_latch_flipped_to_none_is_the_untaken_capsule_byte_for_byte() -> Result<(), &'static str> {
    let flipped = latched_bytes_with_latch(None)?;
    let untaken = live_dcr(SHELL_CONVERSATION_ID, false)?;
    let untaken_bytes = envelope(
        SHELL_CONVERSATION_ID,
        &join(&sections_of(&describe_live(&untaken)?)?)?,
    )?;
    assert_eq!(flipped, untaken_bytes);

    let restored = LifecycleCapsule::decode_canonical(&flipped)
        .map_err(|_| "the flipped capsule decodes")?
        .restore()
        .map_err(|_| "the flipped capsule restores as the untaken state")?;
    let (_, mut frontiers, _) = restored
        .into_parts()
        .lifecycle
        .ok_or("the lifecycle comes back")?
        .into_parts();
    assert_eq!(frontiers, untaken.frontiers);
    frontiers
        .take_fenced_marker_record(untaken.recovery)
        .ok_or("the untaken state grants its one take")?
        .consume();
    Ok(())
}

// ---------------------------------------------------------------------------
// Pin (d): refusals by name
// ---------------------------------------------------------------------------

#[test]
fn every_malformed_lifecycle_section_is_refused_by_name() -> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, false)?;
    let description = describe_live(&live)?;
    let sections = sections_of(&description)?;
    let [participants, frontiers, closure] = sections.as_slice() else {
        return Err("three sections");
    };
    let good = envelope(SHELL_CONVERSATION_ID, &join(&sections)?)?;
    assert!(decode(&good).is_none(), "the hand-built envelope decodes");

    let missing = envelope(
        SHELL_CONVERSATION_ID,
        &join(core::slice::from_ref(participants))?,
    )?;
    assert_eq!(
        decode(&missing),
        Some(CapsuleDecodeError::MissingSection { tag: 3 })
    );

    let swapped = envelope(
        SHELL_CONVERSATION_ID,
        &join(&[participants.clone(), closure.clone(), frontiers.clone()])?,
    )?;
    assert_eq!(
        decode(&swapped),
        Some(CapsuleDecodeError::SectionOutOfOrder { tag: 4 })
    );

    let unknown = envelope(
        SHELL_CONVERSATION_ID,
        &join(&[
            (9, participants.1.clone()),
            frontiers.clone(),
            closure.clone(),
        ])?,
    )?;
    assert_eq!(
        decode(&unknown),
        Some(CapsuleDecodeError::UnknownSection { tag: 9 })
    );

    let mut long_body = participants.1.clone();
    long_body.push(0);
    let declared = u32::try_from(long_body.len()).map_err(|_| "fits")?;
    let long = envelope(
        SHELL_CONVERSATION_ID,
        &join(&[(2, long_body), frontiers.clone(), closure.clone()])?,
    )?;
    assert_eq!(
        decode(&long),
        Some(CapsuleDecodeError::NonCanonicalSectionLength { tag: 2, declared })
    );

    let mut short_body = closure.1.clone();
    short_body.pop();
    let declared = u32::try_from(short_body.len()).map_err(|_| "fits")?;
    let short = envelope(
        SHELL_CONVERSATION_ID,
        &join(&[participants.clone(), frontiers.clone(), (4, short_body)])?,
    )?;
    assert_eq!(
        decode(&short),
        Some(CapsuleDecodeError::NonCanonicalSectionLength { tag: 4, declared })
    );

    Ok(())
}

#[test]
fn every_lifecycle_field_fault_is_refused_by_name() -> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, false)?;
    let sections = sections_of(&describe_live(&live)?)?;
    let [participants, frontiers, closure] = sections.as_slice() else {
        return Err("three sections");
    };

    let bad_closure = envelope(
        SHELL_CONVERSATION_ID,
        &join(&[participants.clone(), frontiers.clone(), (4, vec![9])])?,
    )?;
    assert_eq!(
        decode(&bad_closure),
        Some(CapsuleDecodeError::NonCanonicalSection { tag: 4 })
    );

    // The latch is the frontiers section's last field; with no latch it is a
    // single absent byte, and only 0 or 1 is canonical.
    let mut bad_flag = frontiers.1.clone();
    *bad_flag.last_mut().ok_or("frontiers body")? = 2;
    let bad_latch = envelope(
        SHELL_CONVERSATION_ID,
        &join(&[participants.clone(), (3, bad_flag), closure.clone()])?,
    )?;
    assert_eq!(
        decode(&bad_latch),
        Some(CapsuleDecodeError::NonCanonicalSection { tag: 3 })
    );

    let foreign = envelope(SHELL_CONVERSATION_ID + 1, &join(&sections)?)?;
    assert_eq!(
        decode(&foreign),
        Some(CapsuleDecodeError::ForeignConversation { tag: 3 })
    );
    Ok(())
}

#[test]
fn participants_out_of_order_are_refused_at_capture_and_decode() -> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, false)?;
    let mut doubled = describe_live(&live)?;
    let participant = doubled
        .restore
        .participants
        .first()
        .cloned()
        .ok_or("one participant")?;
    doubled.restore.participants.push(participant);
    let bytes = envelope(
        SHELL_CONVERSATION_ID,
        &encode_description(&doubled).ok_or("the doubled description encodes")?,
    )?;
    assert_eq!(
        decode(&bytes),
        Some(CapsuleDecodeError::ParticipantOutOfOrder {
            participant_id: PARTICIPANT
        })
    );

    let shell = validated_shell(SHELL_CONVERSATION_ID)?;
    let mut view = lifecycle_view(&live);
    let first = view
        .participants
        .first()
        .map(super::ParticipantLifecycleView::participant_id)
        .ok_or("one view")?;
    let again = live
        .participants
        .first()
        .map(RestoredParticipantLifecycle::view)
        .ok_or("one participant")?;
    view.participants.push(again);
    assert_eq!(
        LifecycleCapsule::capture(&LifecycleView {
            conversation: &shell,
            lifecycle: Some(view),
            covered_log_sequence: 0,
        })
        .err(),
        Some(CapsuleCaptureError::ParticipantOutOfOrder {
            participant_id: first
        })
    );
    Ok(())
}

#[test]
fn a_lifecycle_under_another_shell_is_refused_at_capture() -> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, false)?;
    let other_shell = validated_shell(SHELL_CONVERSATION_ID + 1)?;
    assert_eq!(
        capture(&other_shell, &live, 0).err(),
        Some(CapsuleCaptureError::ForeignConversation)
    );
    Ok(())
}

#[test]
fn a_decoded_lifecycle_that_fails_joint_validation_is_refused_at_restore()
-> Result<(), &'static str> {
    let live = live_dcr(SHELL_CONVERSATION_ID, false)?;
    let mut drifted = describe_live(&live)?;
    let Some(ParticipantLifecycleRestore::Live { identity, .. }) =
        drifted.restore.participants.first_mut()
    else {
        return Err("one live participant");
    };
    // The frontier still projects the participant at cursor 0.
    identity.cursor = 5;
    let bytes = envelope(
        SHELL_CONVERSATION_ID,
        &encode_description(&drifted).ok_or("the drifted description encodes")?,
    )?;
    let capsule =
        LifecycleCapsule::decode_canonical(&bytes).map_err(|_| "drift is not a byte fault")?;
    assert!(matches!(
        capsule.restore(),
        Err(CapsuleRestoreError::Lifecycle(_))
    ));
    Ok(())
}