liminal-server 0.20.1

//! Participant operation schema preflight, migration, and cold replay.

use std::collections::VecDeque;

use liminal_protocol::lifecycle::{BindingState, MarkerAnchorReconcile, RecipientAckObligations};
use liminal_protocol::wire::ParticipantDelivery;

use crate::config::types::ParticipantConfig;

use super::barrier::ReceiptCapacityLimits;
use super::fenced_attach_terminal::{AuditSourceError, ComposedTerminalValidation};
use super::log::{
    DecodedOperation, DecodedStoredOperation, OperationLog, OperationLogError,
    OperationSchemaPhase, READ_BATCH_SIZE, SCHEMA_VERSION, SCHEMA_VERSION_V3, StoredDetached,
    StoredDetachedCause, StoredDetachedSource, StoredOperation, StoredOperationV2,
    StoredRetainedCharge, StoredRetainedChargeChange, StoredTerminalDisposition,
    V2AttachedPrestate, encode_v2, encode_versioned, migrate_v2_attached,
};
use super::ops_frontier::stored_retained_charges;
use super::ops_session::DetachReplayInputs;
use super::outbox::ConversationOutboxLimits;
use super::outbox_log::OutboxLog;
use super::outbox_projection::{capture_projection_prestate, project_committed_source};
#[cfg(test)]
use super::outbox_replay::AggregateExtensionMerge;
use super::outbox_replay::{ExtensionMerge, RepairPaging, RestoreError, StagedRepairs};
use super::restore_checkpoint::CheckpointSchema;
use super::restore_checkpoint_locator::CheckpointSource;
use super::restore_from_checkpoint::TailReplayError;
use super::routed_stream::RoutedStreamRefusal;
use super::state::{ConversationAuthority, StateError};

/// Where the single replay pass stands in one conversation's operation log.
///
/// It owns everything the former standalone schema pass owned — the next
/// sequence to read, the schema phase that sequence is read under, and the
/// composed-terminal audit carried across every row already read — so the
/// audit runs on each row the moment it is read, and never on a second read.
pub(super) struct LogWalk {
    next_sequence: u64,
    phase: OperationSchemaPhase,
    exhausted: bool,
    boundary: Option<(OperationLog, u64)>,
    routed_route: Option<String>,
    rebuild_route: Option<String>,
    audit: ComposedTerminalValidation,
    canonical_rows: VecDeque<(u64, Vec<u8>)>,
}

impl LogWalk {
    pub(super) const fn from_checkpoint_start(
        start: super::restore_from_checkpoint::TailStart,
        audit: ComposedTerminalValidation,
    ) -> Self {
        let phase = match start.schema() {
            super::restore_checkpoint::CheckpointSchema::V2Prefix => OperationSchemaPhase::V2Prefix,
            super::restore_checkpoint::CheckpointSchema::V3Suffix => OperationSchemaPhase::V3Suffix,
            super::restore_checkpoint::CheckpointSchema::V4Suffix => OperationSchemaPhase::V4Suffix,
        };
        Self {
            next_sequence: start.operation_sequence(),
            phase,
            exhausted: false,
            boundary: None,
            routed_route: None,
            rebuild_route: None,
            audit,
            canonical_rows: VecDeque::new(),
        }
    }

    fn new(identity_slots: u64) -> Result<Self, StateError> {
        Ok(Self {
            next_sequence: 0,
            phase: OperationSchemaPhase::V2Prefix,
            exhausted: false,
            boundary: None,
            routed_route: None,
            rebuild_route: None,
            audit: ComposedTerminalValidation::for_identity_slots(identity_slots)?,
            canonical_rows: VecDeque::new(),
        })
    }

    /// Carries the audited schema phase into the next checkpoint.
    pub(super) const fn checkpoint_schema(&self) -> CheckpointSchema {
        match self.phase {
            OperationSchemaPhase::V2Prefix => CheckpointSchema::V2Prefix,
            OperationSchemaPhase::V3Suffix => CheckpointSchema::V3Suffix,
            OperationSchemaPhase::V4Suffix => CheckpointSchema::V4Suffix,
        }
    }

    /// Reads to the generation and exclusive head captured at the first page.
    /// A short page advances normally; reaching the head requires no EOF read.
    pub(super) async fn read_page(
        &mut self,
        log: &OperationLog,
    ) -> Result<Option<Vec<DecodedOperation>>, StateError> {
        if self.exhausted {
            return Ok(None);
        }
        if let Some(route) = self.routed_route.as_deref() {
            // A validated boundary with no routed tail behind it has no rows
            // past the boundary, the answer `OperationLog::read_committed_at`
            // gives: a migrated or flat-only stream has a native head and no
            // routed head, and the checkpoint it published at boot covers it.
            let page = match log
                .read_routed_page(route, self.next_sequence, self.phase)
                .await
            {
                Ok(page) => page,
                Err(OperationLogError::RoutedStream(RoutedStreamRefusal::MissingHead {
                    ..
                })) => {
                    self.exhausted = true;
                    return Ok(None);
                }
                Err(error) => return Err(error.into()),
            };
            self.phase = page.next_phase;
            self.canonical_rows = page.canonical_rows;
            if page.rows.is_empty() {
                self.exhausted = true;
                return Ok(None);
            }
            return Ok(Some(page.rows));
        }
        if self.boundary.is_none() {
            let generation = log.generation().await?;
            let pinned =
                OperationLog::at_generation(log.store(), log.conversation_id(), generation);
            let head = match pinned.read_stream_next_seq().await? {
                Some(0) => {
                    return Err(StateError::invariant(
                        "operation stream has a zero stored head",
                    ));
                }
                Some(head) => head,
                None => 0,
            };
            self.boundary = Some((pinned, head));
        }
        let (pinned, head) = self
            .boundary
            .as_ref()
            .ok_or_else(|| StateError::invariant("operation walk has no captured boundary"))?;
        if self.next_sequence >= *head {
            if let Some(route) = self.rebuild_route.as_deref() {
                let page = pinned
                    .read_rebuild_page(route, self.next_sequence, self.phase)
                    .await?;
                if let Some(page) = page {
                    self.phase = page.next_phase;
                    self.canonical_rows = page.canonical_rows;
                    if !page.rows.is_empty() {
                        return Ok(Some(page.rows));
                    }
                }
            } else if self.next_sequence != *head {
                return Err(OperationLogError::Sequence {
                    expected: *head,
                    actual: self.next_sequence,
                }
                .into());
            }
            self.exhausted = true;
            return Ok(None);
        }
        let remaining = head - self.next_sequence;
        let limit = usize::try_from(remaining.min(READ_BATCH_SIZE as u64))
            .map_err(|_| StateError::invariant("operation page limit exceeds address space"))?;
        let page = pinned
            .read_bounded_page(self.next_sequence, self.phase, limit)
            .await?;
        if page.rows.is_empty() {
            return Err(OperationLogError::CorruptRow {
                sequence: self.next_sequence,
            }
            .into());
        }
        if page.rows.len() > limit || page.rows.iter().any(|row| row.sequence >= *head) {
            return Err(StateError::invariant(
                "operation page exceeds its captured head",
            ));
        }
        self.phase = page.next_phase;
        self.canonical_rows = page.canonical_rows;
        Ok(Some(page.rows))
    }

    fn take_canonical(&mut self, sequence: u64) -> Result<Vec<u8>, StateError> {
        let (actual, canonical) = self.canonical_rows.pop_front().ok_or_else(|| {
            StateError::invariant("operation page lost its original source bytes")
        })?;
        if actual != sequence {
            return Err(OperationLogError::Sequence {
                expected: sequence,
                actual,
            }
            .into());
        }
        Ok(canonical)
    }

    /// Audits one decoded row — contiguity, then the composed-terminal schema
    /// audit — and advances past it only when both hold.
    pub(super) async fn audit_row(
        &mut self,
        log: &OperationLog,
        row: &DecodedOperation,
    ) -> Result<(), StateError> {
        if row.sequence != self.next_sequence {
            return Err(OperationLogError::Sequence {
                expected: self.next_sequence,
                actual: row.sequence,
            }
            .into());
        }
        let audited_log = self.boundary.as_ref().map_or(log, |(pinned, _)| pinned);
        if let Some(route) = self.rebuild_route.as_deref() {
            let prefix_head = self
                .boundary
                .as_ref()
                .map(|(_, head)| *head)
                .ok_or_else(|| StateError::invariant("rebuild has no native boundary"))?;
            let read_source = |sequence| async move {
                if sequence < prefix_head {
                    audited_log.read_at(sequence).await
                } else {
                    match audited_log.read_routed_at(route, sequence).await {
                        Err(OperationLogError::RoutedStream(
                            RoutedStreamRefusal::MissingHead { .. },
                        )) => Ok(None),
                        read => read,
                    }
                }
            };
            match self
                .audit
                .validate_from(&read_source, row.sequence, &row.operation)
                .await
            {
                Ok(()) => {}
                Err(AuditSourceError::Semantic(error)) => return Err(error),
                Err(AuditSourceError::Source(error)) => return Err(error.into()),
            }
        } else {
            self.audit
                .validate(audited_log, row.sequence, &row.operation)
                .await?;
        }
        self.next_sequence =
            self.next_sequence
                .checked_add(1)
                .ok_or(StateError::AllocationExhausted {
                    domain: "log sequence",
                })?;
        Ok(())
    }

    async fn audit_row_from<R, F, E>(
        &mut self,
        read_source: &R,
        row: &DecodedOperation,
    ) -> Result<(), AuditSourceError<E>>
    where
        R: Fn(u64) -> F + Sync,
        F: std::future::Future<Output = Result<Option<DecodedOperation>, E>> + Send,
    {
        if row.sequence != self.next_sequence {
            return Err(StateError::from(OperationLogError::Sequence {
                expected: self.next_sequence,
                actual: row.sequence,
            })
            .into());
        }
        self.audit
            .validate_from(read_source, row.sequence, &row.operation)
            .await?;
        self.next_sequence =
            self.next_sequence
                .checked_add(1)
                .ok_or(StateError::AllocationExhausted {
                    domain: "log sequence",
                })?;
        Ok(())
    }

    async fn audit_checkpoint_remaining<R, F, E>(
        &mut self,
        log: &OperationLog,
        read_source: &R,
        unread: std::vec::IntoIter<DecodedOperation>,
    ) -> Result<(), TailReplayError<E>>
    where
        E: std::error::Error + Send + 'static,
        R: Fn(u64) -> F + Sync,
        F: std::future::Future<Output = Result<Option<DecodedOperation>, E>> + Send,
    {
        for row in unread {
            self.audit_row_from(read_source, &row)
                .await
                .map_err(audit_source_error)?;
        }
        while let Some(rows) = self.read_page(log).await? {
            for row in rows {
                self.audit_row_from(read_source, &row)
                    .await
                    .map_err(audit_source_error)?;
            }
        }
        Ok(())
    }

    /// Audits, without applying, the rows of the current page not yet reached
    /// and every page after it. Runs ONLY on the refusal path, after an apply
    /// failure, and reads each row still ahead of the walk exactly once.
    async fn audit_remaining(
        mut self,
        log: &OperationLog,
        unread_rows_of_page: std::vec::IntoIter<DecodedOperation>,
    ) -> Result<(), StateError> {
        for row in unread_rows_of_page {
            self.audit_row(log, &row).await?;
        }
        while let Some(rows) = self.read_page(log).await? {
            for row in rows {
                self.audit_row(log, &row).await?;
            }
        }
        Ok(())
    }
}

fn audit_source_error<E: std::error::Error + 'static>(
    error: AuditSourceError<E>,
) -> TailReplayError<E> {
    match error {
        AuditSourceError::Semantic(error) => error.into(),
        AuditSourceError::Source(error) => TailReplayError::Source(error),
    }
}

#[derive(Clone, Copy)]
enum ReplaySource<'a> {
    Legacy(RepairPaging),
    Rebuild {
        route: &'a str,
        paging: RepairPaging,
    },
}

/// Resolves an apply-phase failure to the refusal the former two-pass replay
/// gave for the same bytes.
///
/// That replay audited the WHOLE log before applying any of it, so an audit
/// failure anywhere won over an apply failure anywhere, and a log that failed
/// its audit was refused with no repair row written. Here an undecodable
/// extension row answers first ([`refusal`]), then the rest of the log is
/// audited; only a log that audits clean end to end answers with the apply
/// failure — after the repair rows staged before it are made durable,
/// exactly the rows the in-line repair had already appended by that point.
///
/// `Ok(None)` when the pass owed more than one page of repairs before the
/// failure and wrote none: the replay runs its second pass, which reaches
/// the same failure with every page written ([`RepairPaging`]).
async fn resolve_apply_failure(
    walk: LogWalk,
    log: &OperationLog,
    unread_rows_of_page: std::vec::IntoIter<DecodedOperation>,
    merge: &mut ExtensionMerge<'_>,
    failure: RestoreError,
) -> Result<
    Option<(
        ConversationAuthority,
        ComposedTerminalValidation,
        CheckpointSchema,
        Option<[u8; 32]>,
    )>,
    RestoreError,
> {
    let failure = refusal(merge, failure).await;
    if matches!(failure, RestoreError::Extension(_)) {
        return Err(failure);
    }
    if let Err(audit_failure) = walk.audit_remaining(log, unread_rows_of_page).await {
        return Err(RestoreError::Semantic(audit_failure));
    }
    match merge.persist_staged_repairs().await? {
        StagedRepairs::Persisted => Err(failure),
        StagedRepairs::Overflowed => Ok(None),
    }
}

/// Answers the refusal the former validate-first extension pass gave ahead of
/// `failure`: the extension fault if the rest of the stream has one, `failure`
/// itself otherwise.
///
/// That pass decoded the whole extension stream before the replay read a base
/// row, so an undecodable extension row anywhere won over every base audit,
/// apply and merge failure, and a stream that did not decode was refused with
/// nothing written. The merge's cursor is finished here instead, so the stream
/// is still read once. A failure that is already the cursor's own is that
/// earliest extension fault and answers as it is.
async fn refusal(merge: &mut ExtensionMerge<'_>, failure: RestoreError) -> RestoreError {
    if matches!(failure, RestoreError::Extension(_)) {
        return failure;
    }
    match merge.validate_rest().await {
        Ok(()) => failure,
        Err(extension) => RestoreError::Extension(extension),
    }
}

impl ConversationAuthority {
    /// Applies only rows beyond the validated checkpoint to its move-owned state.
    pub(super) async fn replay_checkpoint_tail<R, F, E>(
        mut self,
        start: super::restore_from_checkpoint::TailStart,
        saved_audit: (ComposedTerminalValidation, [u8; 32]),
        read_source: R,
        log: &OperationLog,
        outbox_log: &OutboxLog,
        config: &ParticipantConfig,
    ) -> Result<(Self, ComposedTerminalValidation, CheckpointSchema, [u8; 32]), TailReplayError<E>>
    where
        E: std::error::Error + Send + 'static,
        R: Fn(u64) -> F + Send + Sync,
        F: std::future::Future<Output = Result<Option<DecodedOperation>, E>> + Send,
    {
        if self.next_log_sequence != start.operation_sequence()
            || self.conversation_id != log.conversation_id()
        {
            return Err(StateError::invariant("checkpoint owner and tail boundary differ").into());
        }
        let route =
            super::restore_checkpoint_locator::checkpoint_conversation_route(self.conversation_id);
        let outbox = self.outbox.take().ok_or_else(|| {
            StateError::invariant("validated checkpoint has no move-owned outbox")
        })?;
        let mut merge =
            ExtensionMerge::from_checkpoint(outbox_log, &route, start, outbox, saved_audit.1)?;
        let mut walk = start.operation_walk(saved_audit.0);
        walk.routed_route = Some(route);
        if let Err(failure) = merge
            .apply_boundary(&mut self, start.operation_sequence(), None)
            .await
        {
            return Err(refusal(&mut merge, failure).await.into());
        }
        loop {
            let rows = match walk.read_page(log).await {
                Ok(Some(rows)) => rows,
                Ok(None) => break,
                Err(failure) => return Err(refusal(&mut merge, failure.into()).await.into()),
            };
            let mut rows = rows.into_iter();
            while let Some(decoded) = rows.next() {
                if let Err(failure) = walk.audit_row_from(&read_source, &decoded).await {
                    match failure {
                        AuditSourceError::Semantic(error) => {
                            return Err(refusal(&mut merge, error.into()).await.into());
                        }
                        AuditSourceError::Source(error) => {
                            merge
                                .validate_rest()
                                .await
                                .map_err(RestoreError::Extension)?;
                            return Err(TailReplayError::Source(error));
                        }
                    }
                }
                let canonical = walk.take_canonical(decoded.sequence)?;
                merge.observe_source(CheckpointSource::Base, decoded.sequence, &canonical)?;
                if let Err(failure) = self
                    .apply_row(decoded, walk.next_sequence, config, &mut merge, log, None)
                    .await
                {
                    let failure = refusal(&mut merge, failure).await;
                    if matches!(failure, RestoreError::Extension(_)) {
                        return Err(failure.into());
                    }
                    walk.audit_checkpoint_remaining(log, &read_source, rows)
                        .await?;
                    return Err(failure.into());
                }
            }
        }
        merge
            .validate_rest()
            .await
            .map_err(RestoreError::Extension)?;
        if merge.persist_staged_repairs().await? != StagedRepairs::Persisted {
            return Err(StateError::invariant("checkpoint tail staged legacy repairs").into());
        }
        self.validate_replayed_seal()?;
        let prefix = merge
            .prefix_digest()
            .ok_or_else(|| StateError::invariant("checkpoint tail has no prefix witness"))?;
        merge.finish(&mut self, walk.next_sequence)?;
        self.reconcile_load_end_marker_anchors()?;
        let schema = walk.checkpoint_schema();
        Ok((self, walk.audit, schema, prefix))
    }

    /// Cold-replays one conversation's complete durable log in ONE pass.
    ///
    /// Each row is read and decoded once. Its contiguity and composed-terminal
    /// schema audit run first, then the live transition core applies it — the
    /// audit and the apply that used to be two full passes over the log.
    ///
    /// The refusal is the one the two-pass replay gave, for every log:
    ///
    /// - a log whose audit fails at entry N is refused with the error naming
    ///   N. When every earlier entry applied, the audit of N simply runs
    ///   first; when an earlier entry M failed to apply, the rest of the log
    ///   is audited before M's failure is allowed to answer
    ///   ([`resolve_apply_failure`]), because the audit used to run first;
    /// - an extension stream with an undecodable row anywhere is refused by
    ///   that row, ahead of every base failure, and the stream is still read
    ///   once: the merge's cursor is decoded to its end before any refusal
    ///   answers and before any repair is written ([`refusal`]);
    /// - a log refused by its audit gets NO durable write: extension repairs
    ///   are staged in memory and persisted only once the audit has passed the
    ///   whole log (`ExtensionMerge::persist_staged_repairs`);
    /// - staging holds one page of repairs at most. A log that owes more is
    ///   replayed twice: the first pass audits it end to end and writes
    ///   nothing, the second, run only over a log that first pass found
    ///   clean, writes the repairs page by page as they fill
    ///   ([`RepairPaging`]). Every page is one `append_all` and one flush,
    ///   and the rows are byte-identical to the in-line repair's;
    /// - the authority is a private value until the final `Ok`; nothing
    ///   partial is ever handed to the handler.
    ///
    /// This is the legacy flat-stream walk: it reads no routed tail, so it
    /// serves the log migration and the boot-replay measurement, never a live
    /// owner load (`restore_at_boot` is that). When `rewrite` is given, every
    /// row is re-encoded in log order at the current schema: frozen v2 rows
    /// exactly as stored, and every later row in the current grammar with its
    /// retained-charge audit recorded as the change it made, measured on the
    /// replayed state.
    pub(super) async fn replay_rewriting(
        conversation_id: u64,
        log: &OperationLog,
        outbox_log: &OutboxLog,
        config: &ParticipantConfig,
        outbox_limits: ConversationOutboxLimits,
        mut rewrite: Option<&mut LogRewrite>,
    ) -> Result<Self, RestoreError> {
        let rewrite_mark = rewrite.as_deref().map(LogRewrite::mark);
        if let Some((authority, _, _, _)) = Self::replay_pass(
            conversation_id,
            log,
            outbox_log,
            config,
            outbox_limits,
            rewrite.as_deref_mut(),
            ReplaySource::Legacy(RepairPaging::StageOnePage),
        )
        .await?
        {
            return Ok(authority);
        }
        // The first pass audited the whole log clean, owed more than one page
        // of repairs and wrote none of them. The second pass replays the same
        // log from nothing and writes each page as it fills; the first pass's
        // rewritten rows are discarded so each row is recorded once.
        if let (Some(rewrite), Some(mark)) = (rewrite.as_deref_mut(), rewrite_mark) {
            rewrite.rewind(mark);
        }
        Self::replay_pass(
            conversation_id,
            log,
            outbox_log,
            config,
            outbox_limits,
            rewrite,
            ReplaySource::Legacy(RepairPaging::WriteEachPage),
        )
        .await?
        .ok_or_else(|| {
            RestoreError::Semantic(StateError::invariant(
                "Unit 2 extension repairs overflowed a page on the replay pass that writes pages",
            ))
        })
        .map(|(authority, _, _, _)| authority)
    }

    /// Rebuilds complete authority from the native prefix and routed continuation.
    pub(super) async fn replay_for_rebuild(
        conversation_id: u64,
        log: &OperationLog,
        outbox_log: &OutboxLog,
        config: &ParticipantConfig,
        outbox_limits: ConversationOutboxLimits,
    ) -> Result<(Self, ComposedTerminalValidation, CheckpointSchema, [u8; 32]), RestoreError> {
        let route =
            super::restore_checkpoint_locator::checkpoint_conversation_route(conversation_id);
        if let Some(restored) = Self::replay_pass(
            conversation_id,
            log,
            outbox_log,
            config,
            outbox_limits,
            None,
            ReplaySource::Rebuild {
                route: &route,
                paging: RepairPaging::StageOnePage,
            },
        )
        .await?
        {
            let (owner, audit, schema, prefix) = restored;
            return Ok((
                owner,
                audit,
                schema,
                prefix
                    .ok_or_else(|| StateError::invariant("full rebuild has no prefix witness"))?,
            ));
        }
        Self::replay_pass(
            conversation_id,
            log,
            outbox_log,
            config,
            outbox_limits,
            None,
            ReplaySource::Rebuild {
                route: &route,
                paging: RepairPaging::WriteEachPage,
            },
        )
        .await?
        .ok_or_else(|| {
            RestoreError::Semantic(StateError::invariant(
                "full rebuild overflowed its durable repair pass",
            ))
        })
        .and_then(|(owner, audit, schema, prefix)| {
            Ok((
                owner,
                audit,
                schema,
                prefix
                    .ok_or_else(|| StateError::invariant("full rebuild has no prefix witness"))?,
            ))
        })
    }

    /// One replay pass over the whole log under `paging`.
    ///
    /// `Ok(None)` only under [`RepairPaging::StageOnePage`], when the log
    /// audited clean end to end but owed more than one page of repairs: the
    /// pass wrote nothing, and [`Self::replay_rewriting`] runs it again
    /// under [`RepairPaging::WriteEachPage`].
    async fn replay_pass(
        conversation_id: u64,
        log: &OperationLog,
        outbox_log: &OutboxLog,
        config: &ParticipantConfig,
        outbox_limits: ConversationOutboxLimits,
        mut rewrite: Option<&mut LogRewrite>,
        source: ReplaySource<'_>,
    ) -> Result<
        Option<(
            Self,
            ComposedTerminalValidation,
            CheckpointSchema,
            Option<[u8; 32]>,
        )>,
        RestoreError,
    > {
        let mut walk = LogWalk::new(config.identity_slots)?;
        let (route, paging) = match source {
            ReplaySource::Legacy(paging) => (None, paging),
            ReplaySource::Rebuild { route, paging } => {
                walk.rebuild_route = Some(route.to_owned());
                (Some(route), paging)
            }
        };
        let mut authority =
            Self::empty(conversation_id, ReceiptCapacityLimits::from_config(config));
        let prepared_merge = route.map_or_else(
            || ExtensionMerge::new(outbox_log, conversation_id, outbox_limits, paging),
            |route| {
                ExtensionMerge::new_rebuild(
                    outbox_log,
                    route,
                    conversation_id,
                    outbox_limits,
                    paging,
                )
            },
        );
        let mut merge = match prepared_merge {
            Ok(merge) => merge,
            Err(failure) => {
                // No merge cursor exists to finish, so the extension is decoded
                // on its own cursor: its one read on this path.
                let mut cursor = route.map_or_else(
                    || outbox_log.restore_cursor(),
                    |route| outbox_log.restore_rebuild_cursor(route),
                );
                cursor.drain().await.map_err(RestoreError::Extension)?;
                return Err(
                    match walk.audit_remaining(log, Vec::new().into_iter()).await {
                        Ok(()) => RestoreError::Semantic(failure),
                        Err(audit_failure) => RestoreError::Semantic(audit_failure),
                    },
                );
            }
        };
        if let Err(failure) = merge.apply_boundary(&mut authority, 0, None).await {
            return resolve_apply_failure(walk, log, Vec::new().into_iter(), &mut merge, failure)
                .await;
        }
        loop {
            let rows = match walk.read_page(log).await {
                Ok(Some(rows)) => rows,
                Ok(None) => break,
                Err(failure) => return Err(refusal(&mut merge, failure.into()).await),
            };
            let mut rows = rows.into_iter();
            while let Some(decoded) = rows.next() {
                if let Err(failure) = walk.audit_row(log, &decoded).await {
                    return Err(refusal(&mut merge, failure.into()).await);
                }
                let canonical = walk.take_canonical(decoded.sequence)?;
                merge.observe_source(CheckpointSource::Base, decoded.sequence, &canonical)?;
                if let Err(failure) = authority
                    .apply_row(
                        decoded,
                        walk.next_sequence,
                        config,
                        &mut merge,
                        log,
                        rewrite.as_deref_mut(),
                    )
                    .await
                {
                    return resolve_apply_failure(walk, log, rows, &mut merge, failure).await;
                }
            }
        }
        merge
            .validate_rest()
            .await
            .map_err(RestoreError::Extension)?;
        if merge.persist_staged_repairs().await? == StagedRepairs::Overflowed {
            return Ok(None);
        }
        authority.validate_replayed_seal()?;
        let prefix = merge.prefix_digest();
        merge.finish(&mut authority, walk.next_sequence)?;
        authority.reconcile_load_end_marker_anchors()?;
        let schema = walk.checkpoint_schema();
        Ok(Some((authority, walk.audit, schema, prefix)))
    }

    /// Applies one audited row through the exact live transition cores, then
    /// merges the extension rows tied at the boundary after it.
    async fn apply_row(
        &mut self,
        decoded: DecodedOperation,
        boundary: u64,
        config: &ParticipantConfig,
        merge: &mut ExtensionMerge<'_>,
        log: &OperationLog,
        rewrite: Option<&mut LogRewrite>,
    ) -> Result<(), RestoreError> {
        let stored_sequence = decoded.sequence;
        let legacy = decoded.schema_version == SCHEMA_VERSION_V3;
        let frozen_v2 = match (&decoded.operation, rewrite.is_some()) {
            (DecodedStoredOperation::V2(operation), true) => {
                Some(encode_v2(operation).map_err(StateError::from)?)
            }
            _ => None,
        };
        let operation = self.decode_operation(decoded)?;
        let prior_charges = rewrite.as_ref().map(|_| self.stored_frontier_charges());
        if route_in_replay_dispatch(&operation) {
            self.route_fate_occurrence(&operation, stored_sequence)?;
        }
        let operation_for_projection = operation.clone();
        let ack_obligations = match &operation {
            StoredOperation::ZeroDebtAck { request, .. }
            | StoredOperation::NonzeroDebtAck { request, .. } => {
                let acknowledged_through = self
                    .slots
                    .get(&request.participant_id)
                    .map_or(0, |slot| slot.member.cursor());
                Some(
                    merge
                        .recipient_ack_obligations(request.participant_id, acknowledged_through)?,
                )
            }
            _ => None,
        };
        self.begin_observer_progress_source()?;
        let mut facts = capture_projection_prestate(self, &operation_for_projection);
        facts.marker_delivery = self.replay_operation(
            operation,
            stored_sequence,
            config,
            ack_obligations,
            log.store(),
        )?;
        if let Some(rewrite) = rewrite {
            let payload = if let Some(payload) = frozen_v2 {
                payload
            } else {
                let (rewritten, converted) = rewritten_operation(
                    operation_for_projection.clone(),
                    prior_charges.as_deref().unwrap_or_default(),
                    &self.stored_frontier_charges(),
                    stored_sequence,
                )?;
                if converted {
                    rewrite.converted_rows = rewrite.converted_rows.saturating_add(1);
                }
                encode_versioned(&rewritten, stored_sequence, SCHEMA_VERSION)
                    .map_err(StateError::from)?
            };
            if legacy {
                rewrite.legacy_rows = rewrite.legacy_rows.saturating_add(1);
            }
            rewrite.rows.push(payload);
        }
        let expected =
            project_committed_source(self, stored_sequence, &operation_for_projection, facts)?;
        self.end_observer_progress_source()?;
        merge
            .apply_boundary(self, boundary, expected.as_ref())
            .await
    }

    /// The executable frontier's retained charges in their stored form; empty
    /// before the first enrollment.
    fn stored_frontier_charges(&self) -> Vec<StoredRetainedCharge> {
        self.frontier()
            .map(|frontier| stored_retained_charges(frontier.retained_charges()))
            .unwrap_or_default()
    }

    /// F8B R-SEAL (§6.6). Closed carries its own clause: Closed => tokens
    /// empty AND frontier None AND marker set. The clause is a CONJUNCTION
    /// check, not a relaxation — a seal that retired the frontier but kept a
    /// token, or set the marker without retiring the frontier, is refused
    /// here. Bare tokens-empty-with-frontier-Some REMAINS the original refusal
    /// for every shape that is not the sealed one; the corruption tripwire is
    /// not widened.
    fn validate_replayed_seal(&self) -> Result<(), RestoreError> {
        if self.is_closed() {
            if !self.tokens.is_empty() || self.frontier().is_some() {
                return Err(RestoreError::Semantic(StateError::invariant(
                    "sealed conversation retained enrollment tokens or an executable frontier",
                )));
            }
        } else if self.tokens.is_empty() {
            if self.frontier().is_some() {
                return Err(RestoreError::Semantic(StateError::invariant(
                    "durably empty conversation rebuilt an executable frontier",
                )));
            }
        } else if self.frontier().is_none() {
            return Err(RestoreError::Semantic(StateError::invariant(
                "enrolled conversation replay completed without executable frontier ownership",
            )));
        }
        Ok(())
    }

    /// Load-end anchor reconcile: participant erasure or record retirement can
    /// strand a stored marker anchor with NO log row left that could retire it
    /// — replay alone rebuilds the wedge (the conversation-6 residue of the
    /// 2026-08-07 outage). Reconciles toward the census in BOTH directions,
    /// loudly, and names which one it moved.
    ///
    /// ⚠ This runs at the END of the load, which is BEFORE any attach binds a
    /// participant — and binding a participant whose cursor is short of a
    /// retained marker RAISES the census this reconcile just agreed with. No
    /// second reconcile stands after that point, which is why the live
    /// admission arm carries one of its own
    /// (`ConversationAuthority::retry_admission_after_anchor_reconcile`).
    fn reconcile_load_end_marker_anchors(&mut self) -> Result<(), StateError> {
        if self.frontier().is_some() {
            let mut frontier = self.take_frontier()?;
            let outcome = frontier.reconcile_marker_anchor_ledgers();
            self.install_frontier(frontier)?;
            // The "reconciled orphaned marker anchors at load" prefix is the
            // estate's forensics key for this repair and stays byte-identical;
            // the direction is appended, never substituted.
            match outcome {
                MarkerAnchorReconcile::InStep => {}
                MarkerAnchorReconcile::RetiredOrphans(orphaned) => tracing::warn!(
                    conversation_id = self.conversation_id,
                    orphaned,
                    direction = "retired-stored-excess",
                    "reconciled orphaned marker anchors at load -- retired the stored \
                     excess the frontier census can no longer derive"
                ),
                MarkerAnchorReconcile::RemintedDeficit(minted) => tracing::warn!(
                    conversation_id = self.conversation_id,
                    minted,
                    direction = "reminted-census-deficit",
                    "reconciled orphaned marker anchors at load -- re-minted the anchors \
                     the frontier census derives and durable accounting had lost"
                ),
                MarkerAnchorReconcile::Refused { census, stored } => tracing::warn!(
                    conversation_id = self.conversation_id,
                    census,
                    stored,
                    direction = "refused",
                    "reconciled orphaned marker anchors at load -- the closure arithmetic \
                     refused the repair; both ledgers are unchanged"
                ),
            }
        }
        Ok(())
    }

    /// The next page of the reference walk, or `None` once both sources are
    /// read. The flat prefix ends at the native head; the rows a commit
    /// stages routed follow under the conversation's route, read the way a
    /// rebuild reads them. A reference that stopped at the flat prefix saw
    /// none of a routed conversation's rows.
    #[cfg(test)]
    async fn reference_page(
        log: &OperationLog,
        route: &str,
        routed: &mut bool,
        sequence: u64,
        phase: &mut OperationSchemaPhase,
    ) -> Result<Option<super::log::DecodedOperationPage>, StateError> {
        loop {
            let page = if *routed {
                let Some(page) = log.read_rebuild_page(route, sequence, *phase).await? else {
                    return Ok(None);
                };
                page
            } else {
                log.read_bounded_page(sequence, *phase, READ_BATCH_SIZE)
                    .await?
            };
            *phase = page.next_phase;
            if !page.rows.is_empty() {
                return Ok(Some(page));
            }
            if *routed {
                return Ok(None);
            }
            *routed = true;
        }
    }

    /// Frozen pre-W3 complete-vector replay used only by equivalence oracles.
    #[cfg(test)]
    pub(super) async fn replay_aggregate_reference(
        conversation_id: u64,
        log: &OperationLog,
        outbox_log: &OutboxLog,
        extension_rows: Vec<(u64, super::outbox_log::OutboxRow)>,
        config: &ParticipantConfig,
        outbox_limits: ConversationOutboxLimits,
    ) -> Result<Self, StateError> {
        let mut authority =
            Self::empty(conversation_id, ReceiptCapacityLimits::from_config(config));
        let mut merge = AggregateExtensionMerge::new(
            outbox_log,
            extension_rows,
            conversation_id,
            outbox_limits,
        )?;
        validate_operation_schema(log, config.identity_slots).await?;
        merge.apply_boundary(&mut authority, 0, None).await?;
        let mut sequence = 0_u64;
        let mut phase = OperationSchemaPhase::V2Prefix;
        let route =
            super::restore_checkpoint_locator::checkpoint_conversation_route(conversation_id);
        let mut routed = false;
        while let Some(page) =
            Self::reference_page(log, &route, &mut routed, sequence, &mut phase).await?
        {
            let page_len = page.rows.len();
            for decoded in page.rows {
                let stored_sequence = decoded.sequence;
                if stored_sequence != sequence {
                    return Err(StateError::Log(super::log::OperationLogError::Sequence {
                        expected: sequence,
                        actual: stored_sequence,
                    }));
                }
                let operation = authority.decode_operation(decoded)?;
                if route_in_replay_dispatch(&operation) {
                    authority.route_fate_occurrence(&operation, stored_sequence)?;
                }
                let operation_for_projection = operation.clone();
                let ack_obligations = match &operation {
                    StoredOperation::ZeroDebtAck { request, .. }
                    | StoredOperation::NonzeroDebtAck { request, .. } => {
                        let acknowledged_through = authority
                            .slots
                            .get(&request.participant_id)
                            .map_or(0, |slot| slot.member.cursor());
                        Some(merge.recipient_ack_obligations(
                            request.participant_id,
                            acknowledged_through,
                        )?)
                    }
                    _ => None,
                };
                authority.begin_observer_progress_source()?;
                let mut facts = capture_projection_prestate(&authority, &operation_for_projection);
                facts.marker_delivery = authority.replay_operation(
                    operation,
                    stored_sequence,
                    config,
                    ack_obligations,
                    log.store(),
                )?;
                let expected = project_committed_source(
                    &authority,
                    stored_sequence,
                    &operation_for_projection,
                    facts,
                )?;
                authority.end_observer_progress_source()?;
                sequence = sequence
                    .checked_add(1)
                    .ok_or(StateError::AllocationExhausted {
                        domain: "log sequence",
                    })?;
                merge
                    .apply_boundary(&mut authority, sequence, expected.as_ref())
                    .await?;
            }
            if !routed && page_len < READ_BATCH_SIZE {
                routed = true;
            }
        }
        // F8B R-SEAL (§6.6) — the twin of the clause in `replay` above. Both
        // seats carry it because both are load-bearing: the production restore
        // and the equivalence oracle must agree on what a Closed conversation
        // looks like, or the oracle would report drift on every sealed store.
        if authority.is_closed() {
            if !authority.tokens.is_empty() || authority.frontier().is_some() {
                return Err(StateError::invariant(
                    "sealed conversation retained enrollment tokens or an executable frontier",
                ));
            }
        } else if authority.tokens.is_empty() {
            if authority.frontier().is_some() {
                return Err(StateError::invariant(
                    "durably empty conversation rebuilt an executable frontier",
                ));
            }
        } else if authority.frontier().is_none() {
            return Err(StateError::invariant(
                "enrolled conversation replay completed without executable frontier ownership",
            ));
        }
        merge.finish(&mut authority, sequence)?;
        Ok(authority)
    }

    /// Converts a frozen v2 row into the v3 in-memory model using exact replay
    /// prestate where the old Attached option grammar omitted its prior epoch.
    fn decode_operation(&self, decoded: DecodedOperation) -> Result<StoredOperation, StateError> {
        match (decoded.schema_version, decoded.operation) {
            (
                super::log::SCHEMA_VERSION | super::log::SCHEMA_VERSION_V3,
                DecodedStoredOperation::V3(operation),
            ) => Ok(operation),
            (super::log::SCHEMA_VERSION_V2, DecodedStoredOperation::V2(operation)) => {
                self.migrate_v2_operation(operation, decoded.sequence)
            }
            (actual, _) => Err(OperationLogError::SchemaVersion(actual).into()),
        }
    }

    fn migrate_v2_operation(
        &self,
        operation: StoredOperationV2,
        sequence: u64,
    ) -> Result<StoredOperation, StateError> {
        Ok(match operation {
            StoredOperationV2::Genesis { event } => StoredOperation::Genesis { event },
            StoredOperationV2::Enrolled {
                request,
                allocation,
                event,
            } => StoredOperation::Enrolled {
                request,
                allocation,
                event,
            },
            StoredOperationV2::Attached {
                request,
                secret_verified,
                allocation,
                event,
            } => {
                let prestate = match self
                    .slots
                    .get(&request.participant_id)
                    .map(|slot| slot.binding)
                {
                    Some(BindingState::Detached) => V2AttachedPrestate::Detached,
                    Some(BindingState::Bound(active)) => V2AttachedPrestate::Bound {
                        binding_epoch: active.binding_epoch.into(),
                    },
                    Some(BindingState::PendingFinalization(_)) | None => V2AttachedPrestate::Other,
                };
                migrate_v2_attached(
                    request,
                    secret_verified,
                    allocation,
                    event,
                    prestate,
                    sequence,
                )?
            }
            StoredOperationV2::Detached {
                request,
                verifier,
                receiving_epoch,
                terminal_order,
                terminal_seq,
                event,
            } => StoredOperation::Detached {
                row: StoredDetached {
                    participant_id: request.participant_id,
                    binding_epoch: receiving_epoch,
                    cause: StoredDetachedCause::CleanDeregister,
                    terminal_order,
                    disposition: StoredTerminalDisposition::Committed { terminal_seq },
                    source: StoredDetachedSource::ExplicitRequestCommitted {
                        request,
                        secret_verified: true,
                        verifier,
                        receiving_epoch,
                        event,
                    },
                },
            },
            StoredOperationV2::ZeroDebtAck {
                request,
                receiving_epoch,
                contiguously_available_through,
            } => StoredOperation::ZeroDebtAck {
                request,
                receiving_epoch,
                contiguously_available_through,
            },
            StoredOperationV2::MarkerDrained { row } => StoredOperation::MarkerDrained { row },
            StoredOperationV2::RecordAdmission { row } => StoredOperation::RecordAdmission { row },
            StoredOperationV2::Left { row } => StoredOperation::Left { row: row.into() },
        })
    }

    /// Replays one durable entry through the exact live transition cores.
    fn replay_operation(
        &mut self,
        operation: StoredOperation,
        sequence: u64,
        config: &ParticipantConfig,
        ack_obligations: Option<(RecipientAckObligations, u64)>,
        store: std::sync::Arc<dyn liminal::durability::DurableStore>,
    ) -> Result<Option<ParticipantDelivery>, StateError> {
        match operation {
            StoredOperation::Genesis { event } => self.replay_genesis(&event).map(|()| None),
            StoredOperation::Enrolled {
                request,
                allocation,
                event,
            } => self
                .replay_enrolled(request, &allocation, &event, sequence, config)
                .map(|()| None),
            StoredOperation::Attached {
                request,
                secret_verified,
                allocation,
                mode,
                event,
            } => {
                if !secret_verified {
                    return Err(StateError::invariant(
                        "durable attach entry recorded an unverified secret",
                    ));
                }
                self.replay_attached(request, &allocation, &mode, &event, sequence, store)
                    .map(|()| None)
            }
            StoredOperation::Detached { row } => self
                .replay_detached_operation(&row, sequence)
                .map(|()| None),
            StoredOperation::Died { row } => self.replay_died_row(&row, sequence).map(|()| None),
            operation @ (StoredOperation::Ordinary { .. } | StoredOperation::Recovered { .. }) => {
                self.replay_specific_fate(&operation, sequence)
                    .map(|()| None)
            }
            StoredOperation::ZeroDebtAck {
                request,
                receiving_epoch,
                contiguously_available_through,
            } => self
                .replay_zero_debt_ack_row(
                    request,
                    receiving_epoch,
                    contiguously_available_through,
                    ack_obligations,
                )
                .map(|()| None),
            StoredOperation::NonzeroDebtAck {
                request,
                receiving_epoch,
                contiguously_available_through,
                event,
            } => self
                .replay_nonzero_debt_ack(super::ops_nonzero_ack::NonzeroAckReplay {
                    stored_request: request,
                    receiving_epoch,
                    stored_scalar_audit: contiguously_available_through,
                    ack_obligations,
                    event: &event,
                    sequence,
                })
                .map(|()| None),
            StoredOperation::RecordAdmission { row } => self
                .replay_record_admission(&row, sequence, config)
                .map(|()| None),
            StoredOperation::MarkerDrained { row } => {
                self.replay_marker_drain(&row, sequence).map(Some)
            }
            // R18 amendment A7 (§0.18 acceptance 3): the row replays through
            // the same installer the live commit ran, so the replayed store
            // reaches the identical generation and verifier.
            StoredOperation::CredentialReissued { row } => self
                .replay_credential_reissue(&row, sequence)
                .map(|()| None),
            StoredOperation::Left { row } => self.replay_leave(&row).map(|()| None),
        }
    }

    /// Routes one durable Detached row to its exact replay transition by
    /// disposition and source shape: connection-close and explicit sources
    /// replay their fate/detach cores; a committed `Drained` source replays
    /// the candidate-lane terminal drain.
    fn replay_detached_operation(
        &mut self,
        row: &StoredDetached,
        sequence: u64,
    ) -> Result<(), StateError> {
        match (row.disposition, row.source.clone()) {
            (_, StoredDetachedSource::ConnectionClose { .. }) => {
                self.replay_connection_detached(row, sequence)
            }
            (
                StoredTerminalDisposition::Committed { terminal_seq },
                StoredDetachedSource::ExplicitRequestCommitted {
                    request,
                    secret_verified: true,
                    verifier,
                    receiving_epoch,
                    event,
                },
            ) if row.cause == StoredDetachedCause::CleanDeregister
                && row.participant_id == request.participant_id
                && row.binding_epoch == receiving_epoch =>
            {
                self.replay_detached(
                    DetachReplayInputs {
                        request,
                        verifier,
                        receiving_epoch,
                        terminal_order: row.terminal_order,
                        terminal_seq,
                    },
                    &event,
                    sequence,
                )
            }
            (
                StoredTerminalDisposition::Pending,
                StoredDetachedSource::ExplicitRequestPending { .. },
            ) => self.replay_explicit_pending_detached(row, sequence),
            (StoredTerminalDisposition::Committed { .. }, StoredDetachedSource::Drained { .. }) => {
                self.replay_detached_drain_row(row, sequence)
            }
            _ => Err(OperationLogError::CorruptRow { sequence }.into()),
        }
    }
}

#[cfg(test)]
#[derive(Debug, Default, PartialEq, Eq)]
pub(super) struct ValidationMemoryHighWater {
    pub(super) maximum_page_rows: usize,
    pub(super) maximum_active_reservations: usize,
}

/// The former standalone schema-and-contiguity pass, kept ONLY as the
/// two-pass reference the equivalence oracles and the memory-bound pin run
/// against. Production replay no longer calls it: [`ConversationAuthority::replay`]
/// audits each row in the same pass that applies it.
///
/// It walks every committed row where its commit put it, the walk the
/// aggregate reference takes: the flat prefix up to the native head, then
/// the rows a commit stages routed under the conversation's route.
#[cfg(test)]
pub(super) async fn validate_operation_schema(
    log: &OperationLog,
    maximum_active_reservations: u64,
) -> Result<(), StateError> {
    validate_operation_schema_inner(log, maximum_active_reservations, |_, _| {}).await
}

#[cfg(test)]
pub(super) async fn validate_operation_schema_measured(
    log: &OperationLog,
    maximum_active_reservations: u64,
    high_water: &mut ValidationMemoryHighWater,
) -> Result<(), StateError> {
    validate_operation_schema_inner(log, maximum_active_reservations, |page_rows, active| {
        high_water.maximum_page_rows = high_water.maximum_page_rows.max(page_rows);
        high_water.maximum_active_reservations = high_water.maximum_active_reservations.max(active);
    })
    .await
}

#[cfg(test)]
async fn validate_operation_schema_inner(
    log: &OperationLog,
    maximum_active_reservations: u64,
    mut observe_memory: impl FnMut(usize, usize),
) -> Result<(), StateError> {
    let route =
        super::restore_checkpoint_locator::checkpoint_conversation_route(log.conversation_id());
    let mut routed = false;
    let mut sequence = 0_u64;
    let mut phase = OperationSchemaPhase::V2Prefix;
    let mut composed = ComposedTerminalValidation::for_identity_slots(maximum_active_reservations)?;
    while let Some(page) =
        ConversationAuthority::reference_page(log, &route, &mut routed, sequence, &mut phase)
            .await?
    {
        let page_len = page.rows.len();
        observe_memory(page_len, composed.active_reservation_count());
        for decoded in page.rows {
            if decoded.sequence != sequence {
                return Err(OperationLogError::Sequence {
                    expected: sequence,
                    actual: decoded.sequence,
                }
                .into());
            }
            composed.validate(log, sequence, &decoded.operation).await?;
            observe_memory(page_len, composed.active_reservation_count());
            sequence = sequence
                .checked_add(1)
                .ok_or(StateError::AllocationExhausted {
                    domain: "log sequence",
                })?;
        }
        if !routed && page_len < READ_BATCH_SIZE {
            routed = true;
        }
    }
    if sequence == 0 {
        refuse_unread_native_head(log).await?;
    }
    Ok(())
}

/// Refuses a walk that read no row from a log whose native head says rows
/// exist below it: the flat prefix came back empty where the head counts
/// rows, which a pass answering `Ok(())` would read as a durably empty log.
///
/// The routed side needs no second head read here. A walk that read no row
/// asked the routed stream from sequence 0; `read_rebuild_page` answers
/// `None` only when no routed head exists, and a routed head above 0 reads
/// its rows from 0 or refuses the missing one by name.
#[cfg(test)]
async fn refuse_unread_native_head(log: &OperationLog) -> Result<(), StateError> {
    let native_head = log.read_stream_next_seq().await?.unwrap_or(0);
    if native_head > 0 {
        return Err(StateError::invariant(format!(
            "participant conversation {} operation log read no row although its native \
             head is {native_head}",
            log.conversation_id()
        )));
    }
    Ok(())
}

/// Every row of one conversation log re-encoded by
/// [`ConversationAuthority::replay_rewriting`], in log order.
#[derive(Debug, Default)]
pub(super) struct LogRewrite {
    pub(super) rows: Vec<Vec<u8>>,
    /// Rows whose complete retained-charge list became a change.
    pub(super) converted_rows: u64,
    /// Rows stored under schema v3, each re-encoded at the current schema.
    pub(super) legacy_rows: u64,
}

/// How far a [`LogRewrite`] had got: its row count and both counters.
#[derive(Clone, Copy, Debug)]
struct LogRewriteMark {
    rows: usize,
    converted_rows: u64,
    legacy_rows: u64,
}

impl LogRewrite {
    /// Where the rewrite stands now, for [`Self::rewind`].
    fn mark(&self) -> LogRewriteMark {
        LogRewriteMark {
            rows: self.rows.len(),
            converted_rows: self.converted_rows,
            legacy_rows: self.legacy_rows,
        }
    }

    /// Discards everything recorded after `mark`.
    fn rewind(&mut self, mark: LogRewriteMark) {
        self.rows.truncate(mark.rows);
        self.converted_rows = mark.converted_rows;
        self.legacy_rows = mark.legacy_rows;
    }
}

/// One replayed row in the current grammar: a complete retained-charge list
/// becomes the change from `prior` to `resulting`, which replay has already
/// proved equal to that list. Says whether anything was converted.
fn rewritten_operation(
    operation: StoredOperation,
    prior: &[StoredRetainedCharge],
    resulting: &[StoredRetainedCharge],
    sequence: u64,
) -> Result<(StoredOperation, bool), StateError> {
    Ok(match operation {
        StoredOperation::RecordAdmission { mut row } if row.retained_charge_change.is_none() => {
            let complete = row
                .resulting_retained_charges
                .take()
                .ok_or(OperationLogError::CorruptRow { sequence })?;
            if complete != resulting {
                return Err(OperationLogError::CorruptRow { sequence }.into());
            }
            row.retained_charge_change =
                Some(StoredRetainedChargeChange::between(prior, resulting)?);
            (StoredOperation::RecordAdmission { row }, true)
        }
        StoredOperation::MarkerDrained { mut row } if row.retained_charge_change.is_none() => {
            let complete = row
                .resulting_retained_charges
                .take()
                .ok_or(OperationLogError::CorruptRow { sequence })?;
            if complete != resulting {
                return Err(OperationLogError::CorruptRow { sequence }.into());
            }
            row.retained_charge_change =
                Some(StoredRetainedChargeChange::between(prior, resulting)?);
            (StoredOperation::MarkerDrained { row }, true)
        }
        operation => (operation, false),
    })
}

/// Rows whose replay transition does not share its live append core route in
/// the dispatcher. Explicit committed Detached and both specific classes route
/// inside their shared core, exactly once. Drain rows (a `Died` row with
/// `drained: Some`, a `Detached` row with the `Drained` source) route NOWHERE:
/// their pending source row already owns the fate occurrence, and the drain
/// consumed its presentation through `select_finalizer` — exactly as a
/// pending-finalizing `Left` row does.
const fn route_in_replay_dispatch(operation: &StoredOperation) -> bool {
    matches!(
        operation,
        StoredOperation::Died {
            row: super::log::StoredDied { drained: None, .. }
        }
    ) || matches!(
        operation,
        StoredOperation::Detached {
            row: StoredDetached {
                source: StoredDetachedSource::ConnectionClose { .. }
                    | StoredDetachedSource::ExplicitRequestPending { .. },
                ..
            }
        }
    )
}

#[cfg(test)]
mod tail_boundary_tests {
    use std::collections::VecDeque;
    use std::error::Error;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::{Arc, OnceLock};

    use liminal::durability::{
        CountingStore, DurabilityError, DurableStore, ScanCursor, ScanPage, StoreCounts,
        StoredEntry, bridge::block_on,
    };

    use super::super::log::{DecodedOperation, OperationLogError, STREAM_PREFIX};
    use super::super::tests_memory_store::MemoryStore;
    use super::super::tests_seed::Shared;
    use super::{ComposedTerminalValidation, LogWalk, OperationLog, OperationSchemaPhase};
    use super::{SCHEMA_VERSION, StateError, StoredOperation, encode_versioned};

    #[test]
    fn operation_prefix_uses_original_stored_bytes() -> Result<(), Box<dyn Error>> {
        use super::super::restore_checkpoint_locator::{
            CheckpointSource, advance_prefix_digest, checkpoint_prefix_seed,
        };
        let store = Arc::new(MemoryStore::default());
        let encoded = encode_versioned(
            &StoredOperation::Genesis { event: Vec::new() },
            0,
            SCHEMA_VERSION,
        )?;
        let mut stored = encoded.clone();
        stored.extend_from_slice(b" \n");
        block_on(store.append(&format!("{STREAM_PREFIX}7"), stored.clone(), 0))??;
        let log = OperationLog::new(store, 7);
        let mut walk = walk(0)?;
        let rows = block_on(walk.read_page(&log))??.ok_or("missing stored page")?;
        assert_eq!(rows.len(), 1);
        let canonical = walk.take_canonical(rows[0].sequence)?;
        assert_eq!(canonical, stored);
        let digest = advance_prefix_digest(
            checkpoint_prefix_seed(),
            CheckpointSource::Base,
            0,
            &canonical,
        )?;
        assert_ne!(
            digest,
            advance_prefix_digest(
                checkpoint_prefix_seed(),
                CheckpointSource::Base,
                0,
                &encoded
            )?
        );
        assert!(walk.take_canonical(0).is_err());
        Ok(())
    }

    #[test]
    fn extension_prefix_bytes_follow_the_consumed_physical_sequence() -> Result<(), Box<dyn Error>>
    {
        use super::super::outbox_log::{OutboxLog, OutboxRow, encode_row};
        let store = Arc::new(MemoryStore::default());
        let log = OutboxLog::new(store.clone(), 7);
        let row = OutboxRow::AckAdvanced {
            source_log_sequence: 0,
            participant_id: 1,
            through_seq: 0,
        };
        let canonical = encode_row(&row)?;
        block_on(store.append(
            "liminal:participant-production-unit2:7",
            canonical.clone(),
            0,
        ))??;
        let mut cursor = log.restore_cursor();
        assert_eq!(block_on(cursor.front())??.map(|(seq, _)| *seq), Some(0));
        let (sequence, decoded, original) = cursor.pop_front_with_canonical()?;
        assert_eq!(sequence, 0);
        assert_eq!(decoded, row);
        assert_eq!(original, canonical);
        assert!(block_on(cursor.front())??.is_none());
        assert_eq!(cursor.confirmed_head(), Some(1));
        assert!(cursor.pop_front_with_canonical().is_err());
        Ok(())
    }

    const HISTORIES: [u64; 2] = [1_000, 100_000];
    const TAIL: u64 = 3;

    fn fixture() -> Result<Arc<MemoryStore>, Box<dyn Error>> {
        static FIXTURE: OnceLock<Result<Arc<MemoryStore>, String>> = OnceLock::new();
        FIXTURE
            .get_or_init(|| build_fixture().map_err(|error| error.to_string()))
            .as_ref()
            .map(Arc::clone)
            .map_err(|error| error.clone().into())
    }

    fn build_fixture() -> Result<Arc<MemoryStore>, Box<dyn Error>> {
        let store = Arc::new(MemoryStore::default());
        let payload = encode_versioned(
            &StoredOperation::Genesis { event: Vec::new() },
            0,
            SCHEMA_VERSION,
        )?;
        for history in HISTORIES {
            let rows = history
                .checked_add(TAIL)
                .ok_or("tail fixture head overflow")?;
            let payloads = vec![payload.clone(); usize::try_from(rows)?];
            assert_eq!(
                block_on(store.append_all(&format!("{STREAM_PREFIX}{history}"), payloads, 0))??,
                rows,
            );
        }
        Ok(store)
    }

    // This tests the cursor boundary independently of owner/capsule validation.
    // Covered bytes are intentionally never applied as conversation transitions.
    fn walk(next_sequence: u64) -> Result<LogWalk, StateError> {
        Ok(LogWalk {
            next_sequence,
            phase: OperationSchemaPhase::V4Suffix,
            exhausted: false,
            boundary: None,
            routed_route: None,
            rebuild_route: None,
            audit: ComposedTerminalValidation::for_identity_slots(2)?,
            canonical_rows: VecDeque::new(),
        })
    }

    fn drain(walk: &mut LogWalk, log: &OperationLog) -> Result<Vec<u64>, Box<dyn Error>> {
        block_on(async {
            let mut sequences = Vec::new();
            while let Some(rows) = walk.read_page(log).await? {
                for row in rows {
                    walk.audit_row(log, &row).await?;
                    sequences.push(row.sequence);
                }
            }
            Ok(sequences)
        })?
    }

    fn tail_counts(history: u64) -> Result<StoreCounts, Box<dyn Error>> {
        let inner: Arc<dyn DurableStore> = fixture()?;
        let counted = Arc::new(CountingStore::new(Shared(inner)));
        let counts = counted.handle();
        let store: Arc<dyn DurableStore> = counted;
        let log = OperationLog::new(store, history);
        let before = counts.snapshot();
        let expected: Vec<_> = (history..history + TAIL).collect();
        assert_eq!(drain(&mut walk(history)?, &log)?, expected);
        let single = counts.snapshot().since(&before);
        assert_eq!(single.read_from_rows, TAIL);
        assert_eq!(
            single.read_from_calls, 1,
            "the tail must finish without an extra empty read"
        );
        assert_eq!(single.read_at_calls, 0);
        assert_eq!(single.read_value_calls, 1);
        assert_eq!(single.read_stream_next_seq_calls, 1);
        assert_eq!(single.flushes, 0);
        assert_eq!(single.append_calls, 0);
        assert_eq!(single.append_all_calls, 0);
        let before_control = counts.snapshot();
        drain(&mut walk(history)?, &log)?;
        drain(&mut walk(history)?, &log)?;
        let doubled = counts.snapshot().since(&before_control);
        assert_eq!(doubled.read_from_rows, TAIL * 2);
        assert_ne!(
            doubled, single,
            "the pin must reject a doubled-read control"
        );
        let before_control = counts.snapshot();
        drain(&mut walk(0)?, &log)?;
        let ignored = counts.snapshot().since(&before_control);
        assert_eq!(ignored.read_from_rows, history + TAIL);
        assert_ne!(
            ignored, single,
            "the pin must reject an ignored-boundary control"
        );
        Ok(single)
    }

    #[test]
    fn resumed_operation_cursor_reads_equal_tail_at_1000_and_100000() -> Result<(), Box<dyn Error>>
    {
        let short = tail_counts(HISTORIES[0])?;
        let long = tail_counts(HISTORIES[1])?;
        assert_eq!(
            short, long,
            "covered history must add no operation-cursor work"
        );
        println!("operation tail counts at histories {HISTORIES:?}: {short}");
        Ok(())
    }

    #[test]
    fn resumed_audit_refuses_a_gap_at_the_saved_boundary() -> Result<(), Box<dyn Error>> {
        let inner: Arc<dyn DurableStore> = fixture()?;
        let log = OperationLog::new(inner, HISTORIES[0]);
        let mut resumed = walk(HISTORIES[0])?;
        let mut rows = block_on(resumed.read_page(&log))??.ok_or("missing tail page")?;
        let row: DecodedOperation = rows.pop().ok_or("empty tail page")?;
        let refused = block_on(resumed.audit_row(&log, &row))?;
        assert!(matches!(
            refused,
            Err(StateError::Log(OperationLogError::Sequence {
                expected: 1_000,
                actual: 1_002
            }))
        ));
        assert_eq!(
            resumed.next_sequence, HISTORIES[0],
            "a refusal cannot advance the head"
        );
        Ok(())
    }

    #[derive(Debug)]
    struct ShortPageStore {
        inner: Arc<dyn DurableStore>,
        shortened: AtomicBool,
    }

    #[async_trait::async_trait]
    impl DurableStore for ShortPageStore {
        async fn apply_routed_batch(
            &self,
            route: &str,
            items: Vec<liminal::durability::RoutedBatchItem>,
        ) -> Result<(), DurabilityError> {
            self.inner.apply_routed_batch(route, items).await
        }
        async fn get_routed(
            &self,
            route: &str,
            key: &[u8],
        ) -> Result<Option<Vec<u8>>, DurabilityError> {
            self.inner.get_routed(route, key).await
        }
        async fn append(
            &self,
            key: &str,
            payload: Vec<u8>,
            expected: u64,
        ) -> Result<u64, DurabilityError> {
            self.inner.append(key, payload, expected).await
        }
        async fn append_all(
            &self,
            key: &str,
            payloads: Vec<Vec<u8>>,
            expected: u64,
        ) -> Result<u64, DurabilityError> {
            self.inner.append_all(key, payloads, expected).await
        }
        async fn read_from(
            &self,
            key: &str,
            offset: u64,
            limit: usize,
        ) -> Result<Vec<StoredEntry>, DurabilityError> {
            let page = self.inner.read_from(key, offset, limit.min(1)).await?;
            if page.len() == 1 && limit > 1 {
                self.shortened.store(true, Ordering::Relaxed);
            }
            Ok(page)
        }
        async fn read_at(
            &self,
            key: &str,
            sequence: u64,
        ) -> Result<Option<StoredEntry>, DurabilityError> {
            self.inner.read_at(key, sequence).await
        }
        async fn cas(&self, key: &str, old: u64, new: u64) -> Result<(), DurabilityError> {
            self.inner.cas(key, old, new).await
        }
        async fn read_stream_next_seq(
            &self,
            stream_key: &str,
        ) -> Result<Option<u64>, DurabilityError> {
            self.inner.read_stream_next_seq(stream_key).await
        }

        async fn read_value(&self, key: &str) -> Result<Option<u64>, DurabilityError> {
            self.inner.read_value(key).await
        }
        async fn scan_page(
            &self,
            prefix: &str,
            cursor: Option<ScanCursor>,
            limit: usize,
        ) -> Result<ScanPage, DurabilityError> {
            self.inner.scan_page(prefix, cursor, limit).await
        }
        async fn flush(&self) -> Result<(), DurabilityError> {
            self.inner.flush().await
        }
    }

    #[test]
    fn a_short_page_mid_stream_cannot_hide_later_operation_rows() -> Result<(), Box<dyn Error>> {
        let history = HISTORIES[0];
        let inner: Arc<dyn DurableStore> = fixture()?;
        let short = Arc::new(ShortPageStore {
            inner,
            shortened: AtomicBool::new(false),
        });
        let store: Arc<dyn DurableStore> = short.clone();
        let log = OperationLog::new(store, history);
        let rows = drain(&mut walk(history)?, &log)?;
        assert!(
            short.shortened.load(Ordering::Relaxed),
            "the mid-stream short-page premise must be reached"
        );
        assert_eq!(
            rows,
            (history..history + TAIL).collect::<Vec<_>>(),
            "a short page must not hide the durable suffix"
        );
        Ok(())
    }
    struct CursorLive(u64);
    impl super::super::restore_checkpoint::CheckpointLiveState for CursorLive {
        type RestoreContext = u64;
        fn encode_live(
            &self,
        ) -> Result<Vec<u8>, super::super::restore_checkpoint::CheckpointError> {
            Ok(self.0.to_le_bytes().to_vec())
        }
        fn decode_live(
            bytes: &[u8],
            positions: &super::super::restore_checkpoint::CheckpointPositions,
            root: &[u8],
            context: &Self::RestoreContext,
        ) -> Result<Self, super::super::restore_checkpoint::CheckpointError> {
            use super::super::restore_checkpoint::CheckpointError;
            let encoded: [u8; 8] = bytes
                .try_into()
                .map_err(|_| CheckpointError::LiveEncoding)?;
            let live = Self(u64::from_le_bytes(encoded));
            if live.0 != *context {
                return Err(CheckpointError::ForeignConversation {
                    expected: *context,
                    actual: live.0,
                });
            }
            live.validate_live(positions, root)?;
            Ok(live)
        }
        fn validate_live(
            &self,
            positions: &super::super::restore_checkpoint::CheckpointPositions,
            root: &[u8],
        ) -> Result<(), super::super::restore_checkpoint::CheckpointError> {
            use super::super::restore_checkpoint::CheckpointError;
            if positions.conversation_id != self.0 {
                return Err(CheckpointError::ForeignConversation {
                    expected: self.0,
                    actual: positions.conversation_id,
                });
            }
            if root != [1] {
                return Err(CheckpointError::Position {
                    field: "cursor fixture root",
                });
            }
            Ok(())
        }
    }

    fn checkpoint_walk(
        conversation_id: u64,
        next_sequence: u64,
    ) -> Result<LogWalk, Box<dyn Error>> {
        use super::super::restore_checkpoint::{
            CheckpointPositions, CheckpointSchema, RestoreCheckpoint,
        };
        use super::super::restore_from_checkpoint::TailStart;
        if next_sequence == 0 {
            return Ok(LogWalk::new(2)?);
        }
        let positions = CheckpointPositions {
            conversation_id,
            covered_log_sequence: next_sequence - 1,
            next_event_ordinal: 1,
            next_log_sequence: next_sequence,
            next_seq: 1,
            next_order: 1,
            next_participant: 1,
            extension_sequence: 0,
            highest_delivery_seq: 0,
            observer_progress: 0,
            merged_base_sequence: next_sequence,
            merged_extension_sequence: 0,
            schema: CheckpointSchema::V4Suffix,
        };
        let checkpoint = RestoreCheckpoint::new(positions, [1], CursorLive(conversation_id))?;
        Ok(TailStart::from_checkpoint(&checkpoint)
            .operation_walk(ComposedTerminalValidation::for_identity_slots(2)?))
    }

    #[test]
    fn operation_tail_component_is_pinned_by_shared_history_harness() -> Result<(), Box<dyn Error>>
    {
        use crate::boot_history_harness::{
            ReadCounts, ReadSource, StageReads, during_stage, history_pair,
        };
        use std::collections::BTreeMap;

        const ID: u64 = 7;
        const HEAD: &str = "fixture:operation-tail-covered-head";
        const STAGE: &str = "operation-tail-component";
        let stream = format!("{STREAM_PREFIX}{ID}");
        let payload = encode_versioned(
            &StoredOperation::Genesis { event: Vec::new() },
            0,
            SCHEMA_VERSION,
        )?;
        let payload_bytes = u64::try_from(payload.len())?;
        let pair = history_pair("operation-tail-cursor-v1", |store, range| {
            let old_head = block_on(store.read_value(HEAD))??.unwrap_or_default();
            let existing = if old_head == 0 { 0 } else { old_head + TAIL };
            let new_head = range.end;
            let target = new_head
                .checked_add(TAIL)
                .ok_or("tail fixture head overflow")?;
            let count = usize::try_from(
                target
                    .checked_sub(existing)
                    .ok_or("tail fixture head regressed")?,
            )?;
            assert_eq!(
                block_on(store.append_all(&stream, vec![payload.clone(); count], existing))??,
                target
            );
            block_on(store.cas(HEAD, old_head, new_head))??;
            Ok(())
        })?;
        let expected: StageReads = BTreeMap::from([(
            STAGE.to_owned(),
            BTreeMap::from([
                (
                    ReadSource::Value(HEAD.to_owned()),
                    ReadCounts { rows: 1, bytes: 8 },
                ),
                (
                    ReadSource::StreamHead(stream.clone()),
                    ReadCounts { rows: 1, bytes: 8 },
                ),
                (
                    ReadSource::Stream(stream),
                    ReadCounts {
                        rows: TAIL,
                        bytes: TAIL * payload_bytes,
                    },
                ),
            ]),
        )]);
        let measured = pair.assert_stage(STAGE, &expected, |store| {
            during_stage(STAGE, || {
                let head = block_on(store.read_value(HEAD))??.ok_or("tail fixture head absent")?;
                let log = OperationLog::new(store, ID);
                assert_eq!(
                    drain(&mut checkpoint_walk(ID, head)?, &log)?,
                    (head..head + TAIL).collect::<Vec<_>>()
                );
                Ok(())
            })
        })?;
        println!("operation tail measured reads: {measured:?}");
        Ok(())
    }
}