liminal-server 0.23.1

//! Move-only outbox restore, idempotency, retention, and Leave discharge.

use std::error::Error;
use std::io::Write;

use liminal::protocol::{encode, encoded_len};
use liminal_protocol::lifecycle::{
    LiveFrontierOwner, RecipientAckObligations, RecordAdmissionCommit,
};
use liminal_protocol::wire::{
    BindingEpoch, ConnectionIncarnation, Generation, MarkerAck, ParticipantRecord, ServerPush,
};

use crate::server::participant::encode_server_push;

use super::outbox::{ConversationOutbox, ConversationOutboxError, ConversationOutboxLimits};
use super::outbox_log::{
    OutboxRow, ProducedBatch, ProducedSourceKind, ProjectedRecord, StoredMarkerAckCommitted,
};
use super::outbox_projection::ReplayedProjectionFacts;
use super::outbox_replay::ExtensionMerge;

const CONVERSATION: u64 = 0xF0_C3;
const SENDER: u64 = 3;
const RECIPIENT: u64 = 7;

fn test_limits() -> Result<ConversationOutboxLimits, ConversationOutboxError> {
    ConversationOutboxLimits::try_new(64, 64)
}

fn restore_owner(
    rows: Vec<(u64, OutboxRow)>,
) -> Result<ConversationOutbox, ConversationOutboxError> {
    ConversationOutbox::restore(CONVERSATION, rows, test_limits()?)
}

fn ordinary(
    source_sequence: u64,
    delivery_sequence: u64,
    recipients: Vec<u64>,
) -> Result<OutboxRow, ConversationOutboxError> {
    let record = ProjectedRecord::try_new(
        CONVERSATION,
        delivery_sequence,
        ParticipantRecord::OrdinaryRecord {
            sender_participant_id: SENDER,
            payload: vec![1, 2, 3, 4],
        },
        recipients,
        Some(SENDER),
    )?;
    Ok(OutboxRow::Produced(ProducedBatch::new(
        source_sequence,
        ProducedSourceKind::RecordAdmission,
        vec![record],
    )))
}

fn left(
    source_sequence: u64,
    delivery_sequence: u64,
    participant_id: u64,
) -> Result<OutboxRow, ConversationOutboxError> {
    let record = ProjectedRecord::try_new(
        CONVERSATION,
        delivery_sequence,
        ParticipantRecord::Left {
            affected_participant_id: participant_id,
            ended_binding_epoch: None,
        },
        Vec::new(),
        Some(participant_id),
    )?;
    Ok(OutboxRow::Produced(ProducedBatch::new(
        source_sequence,
        ProducedSourceKind::Left,
        vec![record],
    )))
}

#[test]
fn uncertain_duplicate_source_batch_is_idempotent_only_by_exact_bytes() -> Result<(), Box<dyn Error>>
{
    let exact = ordinary(0, 1, vec![RECIPIENT])?;
    let owner = restore_owner(vec![(0, exact.clone()), (1, exact.clone())])?;
    assert_eq!(owner.next_extension_sequence(), 2);
    assert_eq!(owner.source_batch_count(), 1);
    assert_eq!(owner.live_record_count(), 1);

    let OutboxRow::Produced(base_batch) = exact else {
        return Err("ordinary fixture was not Produced".into());
    };
    let mut mutations = Vec::new();

    let mut changed_count = base_batch.clone();
    changed_count.ordered_records.clear();
    mutations.push(OutboxRow::Produced(changed_count));

    let mut changed_recipient = base_batch.clone();
    changed_recipient.ordered_records[0].recipients = vec![RECIPIENT + 1];
    mutations.push(OutboxRow::Produced(changed_recipient));

    let mut changed_body = base_batch.clone();
    changed_body.ordered_records[0].body = ParticipantRecord::OrdinaryRecord {
        sender_participant_id: SENDER,
        payload: vec![9],
    };
    mutations.push(OutboxRow::Produced(changed_body));

    let mut changed_sequence = base_batch.clone();
    changed_sequence.ordered_records[0].delivery_seq = 2;
    mutations.push(OutboxRow::Produced(changed_sequence));

    let mut changed_sender = base_batch.clone();
    changed_sender.ordered_records[0].sender = Some(SENDER + 1);
    mutations.push(OutboxRow::Produced(changed_sender));

    let mut changed_charge = base_batch.clone();
    changed_charge.ordered_records[0].encoded_push_bytes += 1;
    mutations.push(OutboxRow::Produced(changed_charge));

    for mutation in mutations {
        let result = restore_owner(vec![
            (0, OutboxRow::Produced(base_batch.clone())),
            (1, mutation),
        ]);
        assert!(result.is_err(), "conflicting source mutation was accepted");
    }
    Ok(())
}

/// DUR-018 R2: the prune frontier the outbox reports for each recipient is
/// that recipient's own acknowledgement, never past an obligation it still
/// owes; a recipient that never acknowledged stands at 0, and a recipient
/// owed nothing is not reported at all.
#[test]
fn recipient_frontiers_never_pass_an_undelivered_obligation() -> Result<(), Box<dyn Error>> {
    let ack = OutboxRow::AckAdvanced {
        source_log_sequence: 3,
        participant_id: RECIPIENT,
        through_seq: 2,
    };
    let owner = restore_owner(vec![
        (0, ordinary(0, 1, vec![RECIPIENT, RECIPIENT + 1])?),
        (1, ordinary(1, 2, vec![RECIPIENT, RECIPIENT + 1])?),
        (2, ordinary(2, 3, vec![RECIPIENT])?),
        (3, ack),
    ])?;
    // RECIPIENT owes row 3 still: its frontier stays at its acknowledgement,
    // 2, below that obligation. RECIPIENT + 1 never acknowledged: 0.
    assert_eq!(
        owner.recipient_frontiers(),
        vec![(RECIPIENT, 2), (RECIPIENT + 1, 0)]
    );
    assert_eq!(owner.next_live(RECIPIENT), Some(3));
    let acknowledged_all = OutboxRow::AckAdvanced {
        source_log_sequence: 4,
        participant_id: RECIPIENT,
        through_seq: 3,
    };
    let owner = restore_owner(vec![
        (0, ordinary(0, 1, vec![RECIPIENT, RECIPIENT + 1])?),
        (1, ordinary(1, 2, vec![RECIPIENT, RECIPIENT + 1])?),
        (2, ordinary(2, 3, vec![RECIPIENT])?),
        (3, acknowledged_all),
    ])?;
    // Owed nothing, RECIPIENT no longer holds the frontier; RECIPIENT + 1 does.
    assert_eq!(owner.recipient_frontiers(), vec![(RECIPIENT + 1, 0)]);
    Ok(())
}

#[test]
fn cumulative_ack_reclaims_only_the_recipient_prefix() -> Result<(), Box<dyn Error>> {
    let produced = ordinary(0, 1, vec![RECIPIENT, RECIPIENT + 1])?;
    let ack = OutboxRow::AckAdvanced {
        source_log_sequence: 1,
        participant_id: RECIPIENT,
        through_seq: 1,
    };
    let owner = restore_owner(vec![(0, produced), (1, ack)])?;
    assert_eq!(owner.ack_through(RECIPIENT), 1);
    assert_eq!(owner.next_live(RECIPIENT), None);
    assert_eq!(owner.next_live(RECIPIENT + 1), Some(1));
    assert_eq!(owner.live_record_count(), 1);
    assert!(owner.charged_bytes() > 0);
    Ok(())
}

#[test]
fn literal_v2_nonobligation_ack_restores_while_regression_refuses() -> Result<(), Box<dyn Error>> {
    let produced = ordinary(0, 1, vec![RECIPIENT])?;
    let historical = OutboxRow::AckAdvanced {
        source_log_sequence: 1,
        participant_id: RECIPIENT,
        through_seq: 2,
    };
    let historical_owner = restore_owner(vec![(0, produced.clone()), (1, historical)])?;
    assert_eq!(historical_owner.ack_through(RECIPIENT), 2);
    assert_eq!(historical_owner.live_record_count(), 0);

    let ack = OutboxRow::AckAdvanced {
        source_log_sequence: 1,
        participant_id: RECIPIENT,
        through_seq: 1,
    };
    let repeated = OutboxRow::AckAdvanced {
        source_log_sequence: 2,
        participant_id: RECIPIENT,
        through_seq: 1,
    };
    assert!(matches!(
        restore_owner(vec![(0, produced), (1, ack), (2, repeated)]),
        Err(ConversationOutboxError::AckRegression { through_seq: 1, .. })
    ));
    Ok(())
}

fn compacted(
    source_sequence: u64,
    delivery_sequence: u64,
    participant_id: u64,
    abandoned_through: u64,
    recipients: Vec<u64>,
) -> Result<OutboxRow, ConversationOutboxError> {
    let record = ProjectedRecord::try_new(
        CONVERSATION,
        delivery_sequence,
        ParticipantRecord::HistoryCompacted {
            affected_participant_id: participant_id,
            abandoned_after: 0,
            abandoned_through,
            physical_floor_at_decision: abandoned_through,
        },
        recipients,
        None,
    )?;
    Ok(OutboxRow::Produced(ProducedBatch::new(
        source_sequence,
        ProducedSourceKind::MarkerDrained,
        vec![record],
    )))
}

/// A member that never returns is passed by compaction: the records the
/// marker abandoned for it stop being owed to it, so the outbox holds only
/// the marker for it, whatever the history (the door room, 11 October
/// 2026: 94 MB of a 101 MB body held for members that never came back).
/// Its marker stays owed, and every other recipient's obligations are
/// untouched.
#[test]
fn a_compaction_marker_ends_the_abandoned_members_transport_obligation()
-> Result<(), Box<dyn Error>> {
    const RECORDS: u64 = 600;
    const PRESENT: u64 = RECIPIENT + 2;
    let mut rows = Vec::new();
    for sequence in 0..RECORDS {
        rows.push((
            sequence,
            ordinary(sequence, sequence + 1, vec![RECIPIENT, PRESENT])?,
        ));
    }
    let before = restore_owner(rows.clone())?;
    assert_eq!(before.live_record_count(), 600);
    assert_eq!(before.next_live(RECIPIENT), Some(1));

    // PRESENT acknowledges the first 500; compaction passes RECIPIENT
    // through 550.
    rows.push((
        RECORDS,
        OutboxRow::AckAdvanced {
            source_log_sequence: RECORDS,
            participant_id: PRESENT,
            through_seq: 500,
        },
    ));
    rows.push((
        RECORDS + 1,
        compacted(
            RECORDS + 1,
            RECORDS + 1,
            RECIPIENT,
            550,
            vec![RECIPIENT, PRESENT],
        )?,
    ));
    let after = restore_owner(rows.clone())?;
    // Kept: 501..=600, still owed to PRESENT (and 551..=600 to RECIPIENT),
    // and the marker.
    assert_eq!(after.live_record_count(), 101);
    assert_eq!(after.next_live(RECIPIENT), Some(551));
    assert_eq!(after.next_live(PRESENT), Some(501));
    assert!(after.is_marker_obligation(RECIPIENT, RECORDS + 1));
    let census = after.recipient_census();
    assert_eq!(
        census
            .iter()
            .map(|(id, records, _, _)| (*id, *records))
            .collect::<Vec<_>>(),
        vec![(RECIPIENT, 51), (PRESENT, 101)]
    );

    // The same rows restore the same owner: the discharge is a fold of the
    // durable marker row, so replay and checkpoint agree.
    let again = restore_owner(rows)?;
    assert_eq!(again.live_record_count(), after.live_record_count());
    assert_eq!(again.charged_bytes(), after.charged_bytes());
    assert_eq!(
        again.live_recipient_obligation_count(),
        after.live_recipient_obligation_count()
    );
    Ok(())
}

#[test]
fn leave_atomically_discharges_retired_obligations_and_replays() -> Result<(), Box<dyn Error>> {
    let payload = ordinary(0, 1, vec![RECIPIENT])?;
    let leave = left(1, 2, RECIPIENT)?;
    let rows = vec![(0, payload), (1, leave)];
    let first = restore_owner(rows.clone())?;
    // The Leave reclaimed the payload's only record, so only the latest
    // source, the Leave itself, is still held.
    assert_eq!(first.source_batch_count(), 1);
    assert_eq!(first.live_record_count(), 0);
    assert_eq!(first.charged_bytes(), 0);
    assert_eq!(first.next_live(RECIPIENT), None);

    let second = restore_owner(rows)?;
    assert_eq!(second.source_batch_count(), first.source_batch_count());
    assert_eq!(
        second.next_extension_sequence(),
        first.next_extension_sequence()
    );
    assert_eq!(second.charged_bytes(), first.charged_bytes());
    Ok(())
}

#[test]
fn participant_ack_only_advances_receipt_frontier() -> Result<(), Box<dyn Error>> {
    let first = ordinary(0, 1, vec![RECIPIENT])?;
    let third = ordinary(1, 3, vec![RECIPIENT])?;
    let before_rows = vec![(0, first.clone()), (1, third.clone())];
    let before = restore_owner(before_rows.clone())?;
    let before_state = (
        before.ack_through(RECIPIENT),
        before.next_live(RECIPIENT),
        before.live_record_count(),
        before.charged_bytes(),
        before.source_batch_count(),
    );
    assert_eq!(before_state.0, 0);
    assert_eq!(before_state.1, Some(1));

    // Select, encode, enqueue, and write the real production frame. These are
    // transport-progress facts only: rebuilding from the same durable rows must
    // expose byte-for-byte the same head and every receipt-owned fact.
    let offered = before
        .delivery_after(RECIPIENT, 0)
        .ok_or("recipient obligation disappeared before offer")?;
    let frame = encode_server_push(ServerPush::ParticipantDelivery(offered.clone()))
        .map_err(|error| format!("push encoding failed: {error:?}"))?;
    let mut encoded = vec![0; encoded_len(&frame)?];
    let written = encode(&frame, &mut encoded)?;
    encoded.truncate(written);
    let mut socket_bytes = Vec::new();
    socket_bytes.write_all(&encoded)?;
    assert_eq!(socket_bytes, encoded);

    let after_write = restore_owner(before_rows.clone())?;
    assert_eq!(after_write.delivery_after(RECIPIENT, 0), Some(offered));
    assert_eq!(
        (
            after_write.ack_through(RECIPIENT),
            after_write.next_live(RECIPIENT),
            after_write.live_record_count(),
            after_write.charged_bytes(),
            after_write.source_batch_count(),
        ),
        before_state
    );

    // Sequence two is a real conversation gap for this recipient. The exact
    // cumulative endpoint three is nevertheless eligible and releases both
    // obligations through it in one committed AckAdvanced row.
    assert_eq!(
        after_write
            .delivery_after(RECIPIENT, 1)
            .map(|d| d.delivery_seq),
        Some(3)
    );
    let committed = restore_owner(vec![
        (0, first),
        (1, third),
        (
            2,
            OutboxRow::AckAdvanced {
                source_log_sequence: 2,
                participant_id: RECIPIENT,
                through_seq: 3,
            },
        ),
    ])?;
    assert_eq!(committed.ack_through(RECIPIENT), 3);
    assert_eq!(committed.next_live(RECIPIENT), None);
    assert_eq!(committed.live_record_count(), 0);
    assert_eq!(committed.charged_bytes(), 0);

    // Repeated/no-op and regression rows are rejected atomically. A gap endpoint
    // has no source row at all; restoring the unchanged durable prefix retains
    // the original frontier and both obligations.
    for through_seq in [3, 2] {
        let mut rows = before_rows.clone();
        rows.push((
            2,
            OutboxRow::AckAdvanced {
                source_log_sequence: 2,
                participant_id: RECIPIENT,
                through_seq: 3,
            },
        ));
        rows.push((
            3,
            OutboxRow::AckAdvanced {
                source_log_sequence: 3,
                participant_id: RECIPIENT,
                through_seq,
            },
        ));
        assert!(matches!(
            restore_owner(rows),
            Err(ConversationOutboxError::AckRegression { .. })
        ));
    }
    let gap_unchanged = restore_owner(before_rows)?;
    assert_eq!(gap_unchanged.ack_through(RECIPIENT), 0);
    assert_eq!(gap_unchanged.next_live(RECIPIENT), Some(1));
    Ok(())
}

pub(super) fn assert_live_recipient_obligation_bound_holds_without_mutation_and_owner_continues()
-> Result<(), Box<dyn Error>> {
    let limits = ConversationOutboxLimits::try_new(1, 2)?;
    let mut owner = ConversationOutbox::restore(CONVERSATION, Vec::new(), limits)?;
    owner.apply_row(0, ordinary(0, 1, vec![RECIPIENT, RECIPIENT + 1])?)?;
    assert_eq!(owner.live_recipient_obligation_count(), 2);

    let Err(error) = owner.apply_row(1, ordinary(1, 2, vec![RECIPIENT + 2])?) else {
        return Err("third live recipient obligation exceeded the signed bound".into());
    };
    assert!(matches!(
        error,
        ConversationOutboxError::LiveRecipientObligationsExceeded {
            limit: 2,
            attempted: 3
        }
    ));
    assert_eq!(owner.next_extension_sequence(), 1);
    assert_eq!(owner.source_batch_count(), 1);
    assert_eq!(owner.live_recipient_obligation_count(), 2);
    assert!(owner.delivery_after(RECIPIENT, 0).is_some());

    owner.apply_row(
        1,
        OutboxRow::AckAdvanced {
            source_log_sequence: 2,
            participant_id: RECIPIENT,
            through_seq: 1,
        },
    )?;
    assert_eq!(owner.live_recipient_obligation_count(), 1);
    owner.apply_row(2, ordinary(1, 2, vec![RECIPIENT + 2])?)?;
    assert_eq!(owner.live_recipient_obligation_count(), 2);
    assert_eq!(owner.next_extension_sequence(), 3);
    assert!(owner.delivery_after(RECIPIENT + 2, 0).is_some());
    Ok(())
}

#[test]
fn live_recipient_obligation_bound_holds_without_mutation_and_owner_continues()
-> Result<(), Box<dyn Error>> {
    assert_live_recipient_obligation_bound_holds_without_mutation_and_owner_continues()
}

#[test]
fn live_recipient_obligation_bound_rejects_checked_product_overflow() {
    assert!(matches!(
        ConversationOutboxLimits::try_new(u64::MAX, 2),
        Err(ConversationOutboxError::BoundOverflow {
            name: "UNIT2_MAX_LIVE_RECIPIENT_OBLIGATIONS"
        })
    ));
}

#[test]
fn marker_cursor_provenance_reconciles_without_consuming_outbox_state() -> Result<(), Box<dyn Error>>
{
    let marker_cursor = 10;
    let offered_cursor = marker_cursor + 2;
    let binding_epoch = BindingEpoch::new(ConnectionIncarnation::new(4, 2), Generation::ONE);
    let marker = OutboxRow::MarkerAckCommitted(StoredMarkerAckCommitted {
        request: MarkerAck {
            conversation_id: CONVERSATION,
            participant_id: RECIPIENT,
            capability_generation: Generation::ONE,
            marker_delivery_seq: marker_cursor,
        },
        receiving_binding_epoch: binding_epoch,
        offered_marker_delivery_seq: marker_cursor,
        delivered_binding_epoch: binding_epoch,
        from_cursor: 0,
        resulting_cursor: marker_cursor,
        base_log_head: 0,
        extension_sequence: 0,
    });
    let rows = vec![(0, marker)];
    let live = restore_owner(rows.clone())?;
    let cold = restore_owner(rows)?;
    let before = (
        live.next_extension_sequence(),
        live.ack_through(RECIPIENT),
        live.live_record_count(),
        live.live_recipient_obligation_count(),
        live.charged_bytes(),
    );

    assert_eq!(
        live.dispatch_after(RECIPIENT, marker_cursor, Some(offered_cursor))?,
        offered_cursor
    );
    assert_eq!(
        cold.dispatch_after(RECIPIENT, marker_cursor, None)?,
        marker_cursor
    );
    assert!(matches!(
        live.dispatch_after(RECIPIENT, marker_cursor + 1, None),
        Err(ConversationOutboxError::ProtocolCursorProvenance { .. })
    ));
    assert_eq!(
        (
            live.next_extension_sequence(),
            live.ack_through(RECIPIENT),
            live.live_record_count(),
            live.live_recipient_obligation_count(),
            live.charged_bytes(),
        ),
        before
    );
    Ok(())
}

macro_rules! assert_not_impl {
    ($type:ty: $trait:path) => {
        const _: fn() = || {
            struct Probe<T: ?Sized>(core::marker::PhantomData<T>);
            trait AmbiguousIfImplemented<A> {
                fn probe() {}
            }
            impl<T: ?Sized> AmbiguousIfImplemented<()> for Probe<T> {}
            impl<T: ?Sized + $trait> AmbiguousIfImplemented<u8> for Probe<T> {}
            let _ = <Probe<$type> as AmbiguousIfImplemented<_>>::probe;
        };
    };
}

#[test]
fn participant_outbox_owner_is_move_only() {
    crate::server::connection::assert_held_heads_are_move_only();
    assert_not_impl!(ConversationOutbox: Clone);
    assert_not_impl!(ConversationOutbox: Copy);
    assert_not_impl!(LiveFrontierOwner: Clone);
    assert_not_impl!(LiveFrontierOwner: Copy);
    assert_not_impl!(RecipientAckObligations: Clone);
    assert_not_impl!(RecipientAckObligations: Copy);
    assert_not_impl!(RecordAdmissionCommit: Clone);
    assert_not_impl!(RecordAdmissionCommit: Copy);
    assert_not_impl!(ReplayedProjectionFacts: Clone);
    assert_not_impl!(ReplayedProjectionFacts: Copy);
    assert_not_impl!(ExtensionMerge<'static>: Clone);
    assert_not_impl!(ExtensionMerge<'static>: Copy);
}

#[test]
fn checkpoint_outbox_moves_the_saved_owner_and_accepts_only_its_next_tail_row()
-> Result<(), Box<dyn Error>> {
    let produced = ordinary(0, 1, vec![RECIPIENT, RECIPIENT + 1])?;
    let ack = OutboxRow::AckAdvanced {
        source_log_sequence: 1,
        participant_id: RECIPIENT,
        through_seq: 1,
    };
    let owner = restore_owner(vec![(0, produced), (1, ack)])?;
    let measurements = owner.retained_authority_measurements()?;
    let charge = owner.charged_bytes();
    let head = owner.next_extension_sequence();
    let parts = owner.into_restore_parts();
    let exact = format!("{parts:?}");
    let mut restored =
        ConversationOutbox::from_restore_parts(CONVERSATION, head, parts, test_limits()?)?;
    assert_eq!(restored.retained_authority_measurements()?, measurements);
    assert_eq!(restored.charged_bytes(), charge);
    assert_eq!(restored.ack_through(RECIPIENT), 1);
    assert_eq!(restored.next_live(RECIPIENT + 1), Some(1));
    let parts = restored.into_restore_parts();
    assert_eq!(format!("{parts:?}"), exact);
    restored = ConversationOutbox::from_restore_parts(CONVERSATION, head, parts, test_limits()?)?;
    restored.apply_row(head, ordinary(2, 2, vec![RECIPIENT + 1])?)?;
    assert_eq!(restored.next_extension_sequence(), head + 1);
    assert_eq!(restored.live_record_count(), 2);
    assert!(matches!(
        restored.apply_row(head, ordinary(3, 3, vec![RECIPIENT + 1])?),
        Err(ConversationOutboxError::ExtensionSequence { expected, actual })
            if expected == head + 1 && actual == head
    ));
    Ok(())
}

#[test]
fn checkpoint_outbox_refuses_foreign_identity_wrong_saved_head_and_current_limit()
-> Result<(), Box<dyn Error>> {
    for (conversation, head, limit) in [
        (CONVERSATION + 1, 1, test_limits()?),
        (CONVERSATION, 0, test_limits()?),
        (CONVERSATION, 1, ConversationOutboxLimits::try_new(0, 64)?),
    ] {
        let owner = restore_owner(vec![(0, ordinary(0, 1, vec![RECIPIENT])?)])?;
        assert!(
            ConversationOutbox::from_restore_parts(
                conversation,
                head,
                owner.into_restore_parts(),
                limit
            )
            .is_err()
        );
    }
    Ok(())
}

/// The 11 October finding, on a real owner: `RECIPIENT + 1` never
/// acknowledges, and `RECIPIENT` acknowledges everything. Both streams are
/// still cut behind the checkpoint; the records `RECIPIENT + 1` is owed are
/// held by the checkpoint, body and all, and come back from it at boot; and
/// nothing `RECIPIENT` is owed is dropped (it is owed nothing, having
/// acknowledged all).
#[test]
fn a_recipient_that_never_acknowledges_holds_its_records_not_the_cut() -> Result<(), Box<dyn Error>>
{
    let silent = RECIPIENT + 1;
    let owner = restore_owner(vec![
        (0, ordinary(0, 1, vec![RECIPIENT, silent])?),
        (1, ordinary(1, 2, vec![RECIPIENT, silent])?),
        (2, ordinary(2, 3, vec![RECIPIENT, silent])?),
        (
            3,
            OutboxRow::AckAdvanced {
                source_log_sequence: 3,
                participant_id: RECIPIENT,
                through_seq: 3,
            },
        ),
    ])?;
    let held = owner.held_records();
    assert_eq!(
        held,
        super::prune::HeldRecords {
            records: 3,
            obligations: 3,
            lowest_delivery: Some(1),
        }
    );
    let next_sequence = owner.next_extension_sequence();
    let recipients: Vec<_> = owner
        .recipient_frontiers()
        .into_iter()
        .map(
            |(participant_id, acknowledged_through)| super::prune::RecipientAcknowledgement {
                participant_id,
                acknowledged_through,
            },
        )
        .collect();
    assert_eq!(recipients.len(), 1, "only the silent recipient is owed");
    assert_eq!(recipients[0].acknowledged_through, 0);
    let plan = super::prune::plan(
        Some(2),
        CONVERSATION,
        super::prune::StreamBounds::behind_checkpoint(10),
        super::prune::StreamBounds::behind_checkpoint(next_sequence),
        &recipients,
        held,
        0,
    )
    .ok_or("a silent recipient held the cut at 0")?;
    assert_eq!(
        plan.receipts
            .iter()
            .map(|receipt| (receipt.stream, receipt.cut_before))
            .collect::<Vec<_>>(),
        [
            (super::prune::PrunedStream::Operations, 8),
            (super::prune::PrunedStream::Outbox, next_sequence - 2),
        ]
    );
    for receipt in &plan.receipts {
        receipt.prove()?;
        assert_eq!(receipt.held, held);
    }

    // The checkpoint the cut sits behind is the copy that is kept: boot
    // restores the silent recipient's three records from it alone.
    let bytes = owner.encode_checkpoint()?;
    let restored =
        ConversationOutbox::decode_checkpoint(&bytes, CONVERSATION, next_sequence, test_limits()?)?;
    assert_eq!(restored.held_records(), held);
    assert_eq!(restored.live_record_count(), 3);
    assert_eq!(restored.next_live(silent), Some(1));
    assert_eq!(restored.next_live(RECIPIENT), None);
    assert_eq!(restored.ack_through(RECIPIENT), 3);
    Ok(())
}