liminal-server 0.19.1

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

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::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::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.
struct LogWalk {
    next_sequence: u64,
    phase: OperationSchemaPhase,
    exhausted: bool,
    boundary: Option<(OperationLog, u64)>,
    audit: ComposedTerminalValidation,
}

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

    /// Reads to the generation and exclusive head captured at the first page.
    /// A short page advances normally; reaching the head requires no EOF read.
    async fn read_page(
        &mut self,
        log: &OperationLog,
    ) -> Result<Option<Vec<DecodedOperation>>, StateError> {
        if self.exhausted {
            return Ok(None);
        }
        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"))?;
        let remaining =
            head.checked_sub(self.next_sequence)
                .ok_or(OperationLogError::Sequence {
                    expected: *head,
                    actual: self.next_sequence,
                })?;
        if remaining == 0 {
            self.exhausted = true;
            return Ok(None);
        }
        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;
        Ok(Some(page.rows))
    }

    /// Audits one decoded row — contiguity, then the composed-terminal schema
    /// audit — and advances past it only when both hold.
    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);
        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(())
    }

    /// 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(())
    }
}

/// 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>, 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 {
    /// 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.
    pub(super) async fn replay(
        conversation_id: u64,
        log: &OperationLog,
        outbox_log: &OutboxLog,
        config: &ParticipantConfig,
        outbox_limits: ConversationOutboxLimits,
    ) -> Result<Self, RestoreError> {
        Self::replay_rewriting(
            conversation_id,
            log,
            outbox_log,
            config,
            outbox_limits,
            None,
        )
        .await
    }

    /// [`Self::replay`], and when `rewrite` is given, every row 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(),
            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,
            RepairPaging::WriteEachPage,
        )
        .await?
        .ok_or_else(|| {
            RestoreError::Semantic(StateError::invariant(
                "Unit 2 extension repairs overflowed a page on the replay pass that writes pages",
            ))
        })
    }

    /// 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>,
        paging: RepairPaging,
    ) -> Result<Option<Self>, RestoreError> {
        let mut walk = LogWalk::new(config.identity_slots)?;
        let mut authority =
            Self::empty(conversation_id, ReceiptCapacityLimits::from_config(config));
        let mut merge =
            match ExtensionMerge::new(outbox_log, conversation_id, outbox_limits, paging) {
                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.
                    outbox_log
                        .restore_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);
                }
                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()?;
        merge.finish(&mut authority, walk.next_sequence)?;
        authority.reconcile_load_end_marker_anchors()?;
        Ok(Some(authority))
    }

    /// 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(())
    }

    /// 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;
        loop {
            let page = log
                .read_bounded_page(sequence, phase, READ_BATCH_SIZE)
                .await?;
            phase = page.next_phase;
            if page.rows.is_empty() {
                break;
            }
            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 page_len < READ_BATCH_SIZE {
                break;
            }
        }
        // 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.
#[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 mut sequence = 0_u64;
    let mut phase = OperationSchemaPhase::V2Prefix;
    let mut composed = ComposedTerminalValidation::for_identity_slots(maximum_active_reservations)?;
    loop {
        let page = log
            .read_bounded_page(sequence, phase, READ_BATCH_SIZE)
            .await?;
        phase = page.next_phase;
        if page.rows.is_empty() {
            return Ok(());
        }
        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 page_len < READ_BATCH_SIZE {
            return 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::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};

    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,
            audit: ComposedTerminalValidation::for_identity_slots(2)?,
        })
    }

    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 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(())
    }
}