liminal-server 0.20.1

//! Canonical schema-v1 Unit 2 extension stream.
//!
//! This stream is keyed separately from the literal schema-v2 participant
//! operation stream. Its codec is deliberately singular and binary: payload
//! bytes are stored once without JSON/base64 expansion, every enum is explicitly
//! tagged, every length is checked, and trailing or drifted bytes are refused.

mod codec;

#[cfg(test)]
use std::{cell::RefCell, marker::PhantomData};
use std::{collections::VecDeque, sync::Arc};

use liminal::durability::settle::{append_settled, flush_settled};
use liminal::durability::{DurabilityError, DurableStore};
use liminal_protocol::wire::{
    BindingEpoch, MarkerAck, ParticipantDelivery, ParticipantId, ParticipantRecord,
};

use super::facts::{FactsError, now_unix_nanos};
use super::operation_batch::OperationBatch;
use super::restore_checkpoint_locator::checkpoint_conversation_route;
use super::routed_stream::{RoutedStreamError, RoutedStreamRefusal};

pub(super) use codec::{decode_row, encode_row};

/// Stream-key prefix for Unit 2 participant extension rows.
pub(super) const OUTBOX_STREAM_PREFIX: &str = "liminal:participant-production-unit2:";
/// Durable page size signed for Unit 2 restore.
pub(super) const UNIT2_OUTBOX_RESTORE_BATCH_ROWS: usize = 64;
/// Exact Unit 2 extension schema version.
pub(super) const OUTBOX_SCHEMA_VERSION: u8 = 1;

type CanonicalOutboxRow = (u64, OutboxRow, Vec<u8>);

/// Failure to encode, decode, append, or read one Unit 2 extension row.
#[derive(Debug, thiserror::Error)]
pub(super) enum OutboxLogError {
    /// The underlying durable store rejected an operation.
    #[error(transparent)]
    Durability(#[from] DurabilityError),
    /// Routed stream structure or framing was refused.
    #[error("routed stream refused: {0}")]
    RoutedStream(#[source] RoutedStreamRefusal),
    /// The clock could not stamp a row written on its own.
    #[error("Unit 2 extension row timestamp: {0}")]
    Clock(#[from] FactsError),
    /// A row contains no leading schema-version byte.
    #[error("Unit 2 extension row is missing its schema version")]
    MissingSchemaVersion,
    /// A row uses an unsupported schema version.
    #[error("unsupported Unit 2 extension schema version {0}")]
    SchemaVersion(u8),
    /// One stream contains more than one schema version.
    #[error("mixed Unit 2 extension schema versions: expected {expected}, found {actual}")]
    MixedSchemaVersions {
        /// Version established by the first physical row.
        expected: u8,
        /// Different version read later in the same stream.
        actual: u8,
    },
    /// A row ended before its selected schema was complete.
    #[error("Unit 2 extension row ended before {field}")]
    UnexpectedEnd {
        /// Field being decoded when bytes ended.
        field: &'static str,
    },
    /// A numeric tag does not name a schema-v1 value.
    #[error("unknown Unit 2 extension {domain} tag {value}")]
    UnknownTag {
        /// Tagged schema domain.
        domain: &'static str,
        /// Unsupported numeric selector.
        value: u8,
    },
    /// A boolean or optional-value selector was neither zero nor one.
    #[error("invalid Unit 2 extension {field} selector {value}")]
    InvalidSelector {
        /// Selected field.
        field: &'static str,
        /// Invalid selector.
        value: u8,
    },
    /// A persisted capability generation was zero.
    #[error("Unit 2 extension row carries zero capability generation")]
    ZeroGeneration,
    /// A variable-length field exceeds the canonical u32 framing domain.
    #[error("Unit 2 extension {field} length {length} exceeds u32")]
    LengthOverflow {
        /// Variable-length field.
        field: &'static str,
        /// Rejected host length.
        length: usize,
    },
    /// A canonical participant push cannot be measured.
    #[error("Unit 2 extension participant push is not canonically encodable: {0:?}")]
    PushCodec(liminal_protocol::wire::CodecError),
    /// Bytes remain after the one selected schema-v1 row.
    #[error("Unit 2 extension row has {remaining} trailing bytes")]
    TrailingBytes {
        /// Unconsumed bytes.
        remaining: usize,
    },
    /// The durable stream was not contiguous at the expected sequence.
    #[error("Unit 2 extension stream expected sequence {expected}, found {actual}")]
    Sequence {
        /// Next physical sequence required by restore.
        expected: u64,
        /// Sequence supplied by durable storage.
        actual: u64,
    },
    /// The store assigned a different sequence than the optimistic head.
    #[error("Unit 2 extension append expected {expected}, got {actual}")]
    AssignedSequence {
        /// Optimistic-concurrency sequence supplied to the store.
        expected: u64,
        /// Sequence assigned by the store.
        actual: u64,
    },
}

impl From<RoutedStreamError> for OutboxLogError {
    fn from(error: RoutedStreamError) -> Self {
        match error {
            RoutedStreamError::Store(original) => Self::Durability(original),
            RoutedStreamError::Refusal(refusal) => Self::RoutedStream(refusal),
        }
    }
}

/// Exhaustive v2 source kinds that produce a participant delivery batch.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum ProducedSourceKind {
    Enrolled,
    Attached,
    Detached,
    Died,
    MarkerDrained,
    RecordAdmission,
    Left,
}

/// One validated record inside a source batch.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct ProjectedRecord {
    pub(super) delivery_seq: u64,
    pub(super) body: ParticipantRecord,
    pub(super) recipients: Vec<ParticipantId>,
    pub(super) sender: Option<ParticipantId>,
    pub(super) encoded_push_bytes: u64,
}

impl ProjectedRecord {
    /// Builds a projection and records its canonical complete push charge.
    pub(super) fn try_new(
        conversation_id: u64,
        delivery_seq: u64,
        body: ParticipantRecord,
        recipients: Vec<ParticipantId>,
        sender: Option<ParticipantId>,
    ) -> Result<Self, OutboxLogError> {
        let encoded_push_bytes = codec::canonical_push_bytes(conversation_id, delivery_seq, &body)?;
        Ok(Self {
            delivery_seq,
            body,
            recipients,
            sender,
            encoded_push_bytes,
        })
    }

    pub(super) const fn delivery_seq(&self) -> u64 {
        self.delivery_seq
    }

    pub(super) const fn body(&self) -> &ParticipantRecord {
        &self.body
    }

    pub(super) fn recipients(&self) -> &[ParticipantId] {
        &self.recipients
    }

    pub(super) const fn sender(&self) -> Option<ParticipantId> {
        self.sender
    }

    pub(super) const fn encoded_push_bytes(&self) -> u64 {
        self.encoded_push_bytes
    }

    pub(super) fn into_delivery(self, conversation_id: u64) -> ParticipantDelivery {
        ParticipantDelivery {
            conversation_id,
            delivery_seq: self.delivery_seq,
            record: self.body,
        }
    }
}

/// One nonempty, one-or-two-record projection of a committed v2 source.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct ProducedBatch {
    pub(super) source_log_sequence: u64,
    pub(super) source_kind: ProducedSourceKind,
    pub(super) ordered_records: Vec<ProjectedRecord>,
}

impl ProducedBatch {
    pub(super) const fn source_log_sequence(&self) -> u64 {
        self.source_log_sequence
    }

    pub(super) const fn source_kind(&self) -> ProducedSourceKind {
        self.source_kind
    }

    pub(super) fn ordered_records(&self) -> &[ProjectedRecord] {
        &self.ordered_records
    }

    pub(super) const fn new(
        source_log_sequence: u64,
        source_kind: ProducedSourceKind,
        ordered_records: Vec<ProjectedRecord>,
    ) -> Self {
        Self {
            source_log_sequence,
            source_kind,
            ordered_records,
        }
    }
}

/// Complete persistence census for one committed `MarkerAck`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) struct StoredMarkerAckCommitted {
    pub(super) request: MarkerAck,
    pub(super) receiving_binding_epoch: BindingEpoch,
    pub(super) offered_marker_delivery_seq: u64,
    pub(super) delivered_binding_epoch: BindingEpoch,
    pub(super) from_cursor: u64,
    pub(super) resulting_cursor: u64,
    pub(super) base_log_head: u64,
    pub(super) extension_sequence: u64,
}

/// Exhaustive schema-v1 Unit 2 extension row.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum OutboxRow {
    /// One complete source-batch projection.
    Produced(ProducedBatch),
    /// Exact cumulative acknowledgement committed in the v2 stream.
    AckAdvanced {
        source_log_sequence: u64,
        participant_id: ParticipantId,
        through_seq: u64,
    },
    /// Frontier-affecting marker acknowledgement and complete audit.
    MarkerAckCommitted(StoredMarkerAckCommitted),
}

impl OutboxRow {
    /// Returns the row's merge boundary in the literal v2 stream.
    pub(super) const fn base_log_head(&self) -> Option<u64> {
        match self {
            Self::Produced(batch) => batch.source_log_sequence.checked_add(1),
            Self::AckAdvanced {
                source_log_sequence,
                ..
            } => source_log_sequence.checked_add(1),
            Self::MarkerAckCommitted(row) => Some(row.base_log_head),
        }
    }
}

#[cfg(test)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct RestorePageAccounting {
    pub(super) cursors_started: usize,
    pub(super) current_rows: usize,
    pub(super) peak_rows: usize,
    pub(super) pages: usize,
    pub(super) cursor_overlap_observed: bool,
    pub(super) counter_overflow_observed: bool,
}

#[cfg(test)]
#[derive(Default)]
struct ActiveRestorePageAccounting {
    snapshot: RestorePageAccounting,
    active_cursors: usize,
}

#[cfg(test)]
std::thread_local! {
    static RESTORE_PAGE_ACCOUNTING: RefCell<Option<ActiveRestorePageAccounting>> =
        const { RefCell::new(None) };
}

#[cfg(test)]
pub(super) struct RestorePageAccountingGuard {
    _not_send: PhantomData<*const ()>,
    active: bool,
}

#[cfg(test)]
impl RestorePageAccountingGuard {
    pub(super) fn start() -> Self {
        RESTORE_PAGE_ACCOUNTING.with(|accounting| {
            *accounting.borrow_mut() = Some(ActiveRestorePageAccounting::default());
        });
        Self {
            _not_send: PhantomData,
            active: true,
        }
    }

    pub(super) fn snapshot(&self) -> RestorePageAccounting {
        if !self.active {
            return RestorePageAccounting::default();
        }
        RESTORE_PAGE_ACCOUNTING.with(|accounting| {
            accounting
                .borrow()
                .as_ref()
                .map_or_else(RestorePageAccounting::default, |active| active.snapshot)
        })
    }
}

#[cfg(test)]
impl Drop for RestorePageAccountingGuard {
    fn drop(&mut self) {
        self.active = false;
        RESTORE_PAGE_ACCOUNTING.with(|accounting| {
            *accounting.borrow_mut() = None;
        });
    }
}

#[cfg(test)]
fn account_cursor_started() -> bool {
    RESTORE_PAGE_ACCOUNTING.with(|accounting| {
        let mut accounting = accounting.borrow_mut();
        let Some(active) = accounting.as_mut() else {
            return false;
        };
        active.snapshot.cursor_overlap_observed |= active.active_cursors != 0;
        let Some(cursors_started) = active.snapshot.cursors_started.checked_add(1) else {
            active.snapshot.counter_overflow_observed = true;
            return false;
        };
        let Some(active_cursors) = active.active_cursors.checked_add(1) else {
            active.snapshot.counter_overflow_observed = true;
            return false;
        };
        active.snapshot.cursors_started = cursors_started;
        active.active_cursors = active_cursors;
        true
    })
}

#[cfg(test)]
fn account_page_loaded(rows: usize) {
    RESTORE_PAGE_ACCOUNTING.with(|accounting| {
        if let Some(active) = accounting.borrow_mut().as_mut() {
            let snapshot = &mut active.snapshot;
            snapshot.current_rows = rows;
            snapshot.peak_rows = snapshot.peak_rows.max(rows);
            if let Some(next_pages) = snapshot.pages.checked_add(1) {
                snapshot.pages = next_pages;
            } else {
                snapshot.counter_overflow_observed = true;
            }
        }
    });
}

/// Move-only, one-page cursor over one conversation's Unit 2 extension stream.
///
/// A short nonempty store read is never EOF. The cursor advances to the checked
/// successor and confirms EOF only when that offset returns an empty read.
pub(super) struct OutboxRestoreCursor<'a> {
    log: &'a OutboxLog,
    routed_route: Option<String>,
    rebuild_route: Option<String>,
    next_expected_sequence: u64,
    established_version: Option<u8>,
    eof: bool,
    page: VecDeque<(u64, OutboxRow)>,
    canonical_rows: VecDeque<(u64, Vec<u8>)>,
    #[cfg(test)]
    accounted: bool,
}

impl<'a> OutboxRestoreCursor<'a> {
    #[cfg(not(test))]
    const fn new(log: &'a OutboxLog, next_expected_sequence: u64) -> Self {
        Self {
            log,
            routed_route: None,
            rebuild_route: None,
            next_expected_sequence,
            established_version: None,
            eof: false,
            page: VecDeque::new(),
            canonical_rows: VecDeque::new(),
        }
    }

    #[cfg(test)]
    fn new(log: &'a OutboxLog, next_expected_sequence: u64) -> Self {
        Self {
            log,
            routed_route: None,
            rebuild_route: None,
            next_expected_sequence,
            established_version: None,
            eof: false,
            page: VecDeque::new(),
            canonical_rows: VecDeque::new(),
            accounted: account_cursor_started(),
        }
    }

    async fn load_page(&mut self) -> Result<(), OutboxLogError> {
        if self.eof || !self.page.is_empty() {
            return Ok(());
        }
        if let Some(route) = self.routed_route.as_deref() {
            let rows = self
                .log
                .read_routed_page(route, self.next_expected_sequence)
                .await?;
            return self.install_routed_page(rows);
        }
        let entries = self
            .log
            .store
            .read_from(
                &self.log.stream_key,
                self.next_expected_sequence,
                UNIT2_OUTBOX_RESTORE_BATCH_ROWS,
            )
            .await?;
        if entries.is_empty() {
            if let Some(route) = self.rebuild_route.take() {
                let rows = self
                    .log
                    .read_rebuild_page(&route, self.next_expected_sequence)
                    .await?;
                self.routed_route = Some(route);
                return self.install_routed_page(rows.unwrap_or_default());
            }
            self.eof = true;
            return Ok(());
        }

        let mut decoded = VecDeque::with_capacity(entries.len());
        let mut canonical_rows = VecDeque::with_capacity(entries.len());
        for entry in entries {
            if entry.sequence != self.next_expected_sequence {
                return Err(OutboxLogError::Sequence {
                    expected: self.next_expected_sequence,
                    actual: entry.sequence,
                });
            }
            let version = entry
                .payload
                .first()
                .copied()
                .ok_or(OutboxLogError::MissingSchemaVersion)?;
            if let Some(expected) = self.established_version {
                if version != expected {
                    return Err(OutboxLogError::MixedSchemaVersions {
                        expected,
                        actual: version,
                    });
                }
            } else {
                self.established_version = Some(version);
            }
            if version != OUTBOX_SCHEMA_VERSION {
                return Err(OutboxLogError::SchemaVersion(version));
            }
            decoded.push_back((self.next_expected_sequence, decode_row(&entry.payload)?));
            canonical_rows.push_back((entry.sequence, entry.payload));
            self.next_expected_sequence =
                self.next_expected_sequence
                    .checked_add(1)
                    .ok_or(OutboxLogError::Sequence {
                        expected: u64::MAX,
                        actual: entry.sequence,
                    })?;
        }
        self.page = decoded;
        self.canonical_rows = canonical_rows;
        #[cfg(test)]
        if self.accounted {
            account_page_loaded(self.page.len());
        }
        Ok(())
    }

    fn install_routed_page(&mut self, rows: Vec<CanonicalOutboxRow>) -> Result<(), OutboxLogError> {
        if rows.is_empty() {
            self.eof = true;
            return Ok(());
        }
        let mut page = VecDeque::with_capacity(rows.len());
        let mut canonical_rows = VecDeque::with_capacity(rows.len());
        for (sequence, row, canonical) in rows {
            if sequence != self.next_expected_sequence {
                return Err(OutboxLogError::Sequence {
                    expected: self.next_expected_sequence,
                    actual: sequence,
                });
            }
            self.next_expected_sequence =
                self.next_expected_sequence
                    .checked_add(1)
                    .ok_or(OutboxLogError::Sequence {
                        expected: u64::MAX,
                        actual: sequence,
                    })?;
            page.push_back((sequence, row));
            canonical_rows.push_back((sequence, canonical));
        }
        self.page = page;
        self.canonical_rows = canonical_rows;
        #[cfg(test)]
        if self.accounted {
            account_page_loaded(self.page.len());
        }
        Ok(())
    }

    /// Borrows the current physical head, loading exactly one page when needed.
    pub(super) async fn front(&mut self) -> Result<Option<&(u64, OutboxRow)>, OutboxLogError> {
        self.load_page().await?;
        Ok(self.page.front())
    }

    /// Consumes the current physical head after a successful [`Self::front`].
    pub(super) fn pop_front(&mut self) -> Option<(u64, OutboxRow)> {
        let row = self.page.pop_front();
        self.canonical_rows.pop_front();
        #[cfg(test)]
        if row.is_some() {
            self.account_current_rows(self.page.len());
        }
        row
    }

    /// Moves the original stored bytes with the row consumed by the merge.
    pub(super) fn pop_front_with_canonical(
        &mut self,
    ) -> Result<CanonicalOutboxRow, OutboxLogError> {
        let (sequence, _) = self
            .canonical_rows
            .front()
            .ok_or(OutboxLogError::MissingSchemaVersion)?;
        let (row_sequence, _) = self
            .page
            .front()
            .ok_or(OutboxLogError::MissingSchemaVersion)?;
        if sequence != row_sequence {
            return Err(OutboxLogError::Sequence {
                expected: *row_sequence,
                actual: *sequence,
            });
        }
        let (_, canonical) = self
            .canonical_rows
            .pop_front()
            .ok_or(OutboxLogError::MissingSchemaVersion)?;
        let (sequence, row) = self
            .page
            .pop_front()
            .ok_or(OutboxLogError::MissingSchemaVersion)?;
        #[cfg(test)]
        self.account_current_rows(self.page.len());
        Ok((sequence, row, canonical))
    }

    #[cfg(test)]
    fn account_current_rows(&self, rows: usize) {
        if !self.accounted {
            return;
        }
        RESTORE_PAGE_ACCOUNTING.with(|accounting| {
            if let Some(active) = accounting.borrow_mut().as_mut() {
                active.snapshot.current_rows = rows;
            }
        });
    }

    /// Decodes and discards every row not yet consumed, through explicit
    /// empty-read EOF, one page at a time.
    pub(super) async fn drain(&mut self) -> Result<(), OutboxLogError> {
        while self.front().await?.is_some() {
            self.pop_front();
        }
        Ok(())
    }

    /// Returns the checked successor offset after explicit EOF confirmation.
    pub(super) const fn confirmed_head(&self) -> Option<u64> {
        if self.eof {
            Some(self.next_expected_sequence)
        } else {
            None
        }
    }
}

#[cfg(test)]
impl Drop for OutboxRestoreCursor<'_> {
    fn drop(&mut self) {
        self.account_current_rows(0);
        if self.accounted {
            RESTORE_PAGE_ACCOUNTING.with(|accounting| {
                if let Some(active) = accounting.borrow_mut().as_mut() {
                    if let Some(active_cursors) = active.active_cursors.checked_sub(1) {
                        active.active_cursors = active_cursors;
                    } else {
                        active.snapshot.counter_overflow_observed = true;
                    }
                }
            });
        }
    }
}

/// Append-only handle over one conversation's Unit 2 extension stream.
#[derive(Debug)]
pub(super) struct OutboxLog {
    store: Arc<dyn DurableStore>,
    conversation_id: u64,
    stream_key: String,
}

impl OutboxLog {
    pub(super) fn new(store: Arc<dyn DurableStore>, conversation_id: u64) -> Self {
        Self {
            store,
            conversation_id,
            stream_key: format!("{OUTBOX_STREAM_PREFIX}{conversation_id}"),
        }
    }

    /// Starts a fresh bounded traversal at physical sequence zero.
    #[cfg(not(test))]
    pub(super) const fn restore_cursor(&self) -> OutboxRestoreCursor<'_> {
        OutboxRestoreCursor::new(self, 0)
    }

    /// Starts a fresh instrumented bounded traversal in tests.
    #[cfg(test)]
    pub(super) fn restore_cursor(&self) -> OutboxRestoreCursor<'_> {
        OutboxRestoreCursor::new(self, 0)
    }

    /// Resumes at the validated checkpoint's exclusive extension head.
    #[cfg(not(test))]
    pub(super) const fn restore_cursor_from(&self, sequence: u64) -> OutboxRestoreCursor<'_> {
        OutboxRestoreCursor::new(self, sequence)
    }
    #[cfg(test)]
    pub(super) fn restore_cursor_from(&self, sequence: u64) -> OutboxRestoreCursor<'_> {
        OutboxRestoreCursor::new(self, sequence)
    }

    /// Resumes the validated extension boundary on the routed atomic stream.
    pub(super) fn restore_routed_cursor_from(
        &self,
        route: &str,
        sequence: u64,
    ) -> OutboxRestoreCursor<'_> {
        let mut cursor = self.restore_cursor_from(sequence);
        cursor.routed_route = Some(route.to_owned());
        cursor
    }

    /// Rebuilds the native prefix and any explicitly routed continuation.
    pub(super) fn restore_rebuild_cursor(&self, route: &str) -> OutboxRestoreCursor<'_> {
        let mut cursor = OutboxRestoreCursor::new(self, 0);
        cursor.rebuild_route = Some(route.to_owned());
        cursor
    }

    /// Stages one canonical extension row into its source operation's batch.
    ///
    /// # Errors
    /// Propagates codec, store and conditional-head refusals unchanged.
    pub(super) async fn stage_into(
        &self,
        batch: &mut OperationBatch,
        row: &OutboxRow,
        expected_sequence: u64,
        timestamp: u64,
    ) -> Result<(), OutboxLogError> {
        let result: Result<(), OutboxLogError> = async {
            let payload = encode_row(row)?;
            batch
                .stage_source_stream(
                    self.store.as_ref(),
                    &self.stream_key,
                    super::restore_checkpoint_locator::CheckpointSource::Extension,
                    liminal::durability::StoredEntry {
                        payload,
                        sequence: expected_sequence,
                        timestamp,
                    },
                )
                .await?;
            Ok(())
        }
        .await;
        if result.is_err() {
            batch.refuse();
        }
        result
    }

    /// Reads only the routed extension tail after a validated checkpoint boundary.
    ///
    /// # Errors
    /// Refuses a missing or malformed head, missing row and invalid row codec.
    pub(super) async fn read_routed_page(
        &self,
        route: &str,
        from_sequence: u64,
    ) -> Result<Vec<CanonicalOutboxRow>, OutboxLogError> {
        let head = super::routed_stream::read_head(self.store.as_ref(), route, &self.stream_key)
            .await?
            .ok_or_else(|| {
                OutboxLogError::RoutedStream(RoutedStreamRefusal::MissingHead {
                    stream: self.stream_key.as_bytes().to_vec(),
                })
            })?;
        let entries = super::routed_stream::read_page(
            self.store.as_ref(),
            route,
            &self.stream_key,
            &head,
            from_sequence,
            UNIT2_OUTBOX_RESTORE_BATCH_ROWS,
        )
        .await?;
        entries
            .into_iter()
            .map(|entry| Ok((entry.sequence, decode_row(&entry.payload)?, entry.payload)))
            .collect()
    }

    /// Reads an existing routed continuation during an explicit full rebuild.
    pub(super) async fn read_rebuild_page(
        &self,
        route: &str,
        from_sequence: u64,
    ) -> Result<Option<Vec<CanonicalOutboxRow>>, OutboxLogError> {
        let Some(head) =
            super::routed_stream::read_head(self.store.as_ref(), route, &self.stream_key).await?
        else {
            return Ok(None);
        };
        let entries = super::routed_stream::read_page(
            self.store.as_ref(),
            route,
            &self.stream_key,
            &head,
            from_sequence,
            UNIT2_OUTBOX_RESTORE_BATCH_ROWS,
        )
        .await?;
        entries
            .into_iter()
            .map(|entry| Ok((entry.sequence, decode_row(&entry.payload)?, entry.payload)))
            .collect::<Result<Vec<_>, _>>()
            .map(Some)
    }

    /// Stages the replay-validated extension head with the first checkpoint.
    ///
    /// # Errors
    /// Propagates storage and conditional-head refusals unchanged.
    pub(super) async fn stage_bootstrap_head(
        &self,
        batch: &mut OperationBatch,
        next_sequence: u64,
    ) -> Result<(), OutboxLogError> {
        batch
            .stage_bootstrap_stream(self.store.as_ref(), &self.stream_key, next_sequence)
            .await?;
        Ok(())
    }

    /// Appends one canonical row at the exact optimistic head as one routed
    /// durable operation: a one-row batch on the conversation's route,
    /// committed through the path the handler commits every operation.
    ///
    /// The handler stages its extension rows into the operation's own
    /// batch; this is the append for a row written on its own (a marker ack
    /// committed outside an operation, a test-local appender). A flat native
    /// row beside routed rows leaves the native head behind the routed one,
    /// so a row of a routed conversation is never written flat.
    ///
    /// # Errors
    /// Propagates codec, head and store refusals unchanged.
    pub(super) async fn append(
        &self,
        row: &OutboxRow,
        expected_sequence: u64,
    ) -> Result<(), OutboxLogError> {
        self.append_page(std::slice::from_ref(row), expected_sequence)
            .await
    }

    /// Appends one flat native row at the exact optimistic head, then
    /// flushes: the row shape a store written before pruning holds, kept for
    /// the legacy-rows arm alone.
    pub(super) async fn append_flat(
        &self,
        row: &OutboxRow,
        expected_sequence: u64,
    ) -> Result<(), OutboxLogError> {
        let payload = encode_row(row)?;
        let assigned = append_settled(
            self.store.as_ref(),
            &self.stream_key,
            payload,
            expected_sequence,
        )
        .await?;
        if assigned != expected_sequence {
            return Err(OutboxLogError::AssignedSequence {
                expected: expected_sequence,
                actual: assigned,
            });
        }
        flush_settled(self.store.as_ref()).await?;
        Ok(())
    }

    /// Appends contiguous canonical rows starting at the exact optimistic head
    /// as one routed durable operation: every row of the page staged into one
    /// batch on the conversation's route and committed once, never row by
    /// row.
    ///
    /// # Errors
    /// Propagates codec, head and store refusals unchanged.
    pub(super) async fn append_page(
        &self,
        rows: &[OutboxRow],
        expected_sequence: u64,
    ) -> Result<(), OutboxLogError> {
        if rows.is_empty() {
            return Ok(());
        }
        let timestamp = now_unix_nanos()?;
        let mut batch = OperationBatch::new(&checkpoint_conversation_route(self.conversation_id));
        let mut sequence = expected_sequence;
        for row in rows {
            self.stage_into(&mut batch, row, sequence, timestamp)
                .await?;
            sequence = sequence
                .checked_add(1)
                .ok_or(OutboxLogError::AssignedSequence {
                    expected: expected_sequence,
                    actual: u64::MAX,
                })?;
        }
        batch.commit(self.store.as_ref(), || {}).await?;
        Ok(())
    }

    /// Every committed extension row in order: the flat native prefix read
    /// as the frozen pre-W3 aggregate reference (the old short-page
    /// termination and its own decode loop), then the rows a commit stages
    /// routed, at and after the native head under the conversation's route.
    /// A reader of the flat stream alone saw none of the routed rows.
    /// Production restore has no selector or fallback to this implementation.
    #[cfg(test)]
    pub(super) async fn read_all(&self) -> Result<Vec<(u64, OutboxRow)>, OutboxLogError> {
        let mut rows = Vec::new();
        let mut sequence = 0_u64;
        let mut established_version = None;
        loop {
            let entries = self
                .store
                .read_from(&self.stream_key, sequence, UNIT2_OUTBOX_RESTORE_BATCH_ROWS)
                .await?;
            if entries.is_empty() {
                break;
            }
            let page_len = entries.len();
            for entry in entries {
                if entry.sequence != sequence {
                    return Err(OutboxLogError::Sequence {
                        expected: sequence,
                        actual: entry.sequence,
                    });
                }
                let version = entry
                    .payload
                    .first()
                    .copied()
                    .ok_or(OutboxLogError::MissingSchemaVersion)?;
                if let Some(expected) = established_version {
                    if version != expected {
                        return Err(OutboxLogError::MixedSchemaVersions {
                            expected,
                            actual: version,
                        });
                    }
                } else {
                    established_version = Some(version);
                }
                if version != OUTBOX_SCHEMA_VERSION {
                    return Err(OutboxLogError::SchemaVersion(version));
                }
                rows.push((sequence, decode_row(&entry.payload)?));
                sequence = sequence.checked_add(1).ok_or(OutboxLogError::Sequence {
                    expected: u64::MAX,
                    actual: entry.sequence,
                })?;
            }
            if page_len < UNIT2_OUTBOX_RESTORE_BATCH_ROWS {
                break;
            }
        }
        let route = checkpoint_conversation_route(self.conversation_id);
        loop {
            let Some(page) = self.read_rebuild_page(&route, sequence).await? else {
                break;
            };
            if page.is_empty() {
                break;
            }
            for (actual, row, _) in page {
                if actual != sequence {
                    return Err(OutboxLogError::Sequence {
                        expected: sequence,
                        actual,
                    });
                }
                rows.push((sequence, row));
                sequence = sequence.checked_add(1).ok_or(OutboxLogError::Sequence {
                    expected: u64::MAX,
                    actual,
                })?;
            }
        }
        Ok(rows)
    }
}