liminal-server 0.14.0

//! Ordinary record admission, marker drain, and their exact cold replay.
//!
//! Every authorized transition temporarily consumes the conversation's
//! validated live frontier owner. Marker-drain and record commits cross one
//! append/flush boundary before the replacement owner, causal counters, or
//! response become observable. Refusals remain protocol-selected and return
//! the complete unchanged owner.

use liminal_protocol::algebra::ResourceVector;
use liminal_protocol::lifecycle::{
    BindingState, CapacityCounter, ConnectionConversationTracking, ImmutableSequenceCandidate,
    LiveFrontierOwner, MarkerAnchorReconcile, MarkerDeliveryProjection, OrdinaryProjectionError,
    PresentedIdentity, RecordAdmissionCommit, RecordAdmissionDecision, RecordAdmissionFailure,
    RecordAdmissionFault, RecordAdmissionPrestate, RetainedRecordCharge,
    SemanticConnectionCapacityDecision, apply_record_admission as select_record_admission,
    classify_record_admission_binding, drain_next_marker,
};
use liminal_protocol::wire::{
    BindingEpoch, DeliverySeq, ParticipantDelivery, ParticipantId, RecordAdmission,
    RecordAdmissionResponse, RecordCommitted,
};

use crate::config::types::ParticipantConfig;
use crate::server::participant::dispatch_impact::DispatchImpactAccumulator;

use super::barrier::{ArmOutcome, OperationFacts};
use super::facts::{Digest, ordinary_payload_fingerprint};
use super::frontier::{ordinary_projection_limits, ordinary_record_charge};
use super::log::{
    StoredBindingEpoch, StoredMarkerDrain, StoredOperation, StoredRecordAdmission,
    StoredRecordAdmissionRequest, StoredResourceVector, StoredRetainedCharge,
};
use super::outbox_projection::ReplayedProjectionFacts;
use super::state::{CommittedAdmissionKey, ConversationAuthority, DurableAppend, StateError};

/// What one census reconcile plus one re-presented admission decided.
///
/// Two arms and no third: either the repaired ledger let the SAME admission
/// commit, or it did not and the caller's already-minted terminal refusal is
/// the answer. `Unchanged` still carries the reconciled owner — the repair is
/// kept either way (see
/// [`ConversationAuthority::retry_admission_after_anchor_reconcile`]).
enum LiveAnchorRetry {
    /// The re-presented admission committed; publish it as any commit.
    Committed(Box<RecordAdmissionCommit>),
    /// No commit. Install this owner and answer the terminal refusal.
    Unchanged(Box<LiveFrontierOwner>),
}

/// How a record-admission protocol fault is to be answered.
///
/// Both arms cross one `&mut self` boundary in the caller and neither borrows
/// the aggregate that produced it, which is the whole reason this type exists:
/// the fault answer is decided under `&self` (the total selector holds a borrow
/// of the participant slot) and published under `&mut self`.
enum RecordAdmissionFaultAnswer {
    /// The fault was census-repairable and the re-presented admission committed.
    Committed(Box<RecordAdmissionCommit>),
    /// Install this owner, then answer this terminal typed refusal.
    Terminal {
        response: RecordAdmissionResponse,
        owner: Box<LiveFrontierOwner>,
    },
}

impl ConversationAuthority {
    #[cfg(test)]
    pub(super) fn apply_record_admission(
        &mut self,
        request: &RecordAdmission,
        operation_facts: &OperationFacts,
        config: &ParticipantConfig,
        appender: &dyn DurableAppend,
    ) -> Result<ArmOutcome, StateError> {
        let mut impact = DispatchImpactAccumulator::new();
        self.apply_record_admission_with_impact(
            request,
            operation_facts,
            config,
            appender,
            &mut impact,
        )
    }

    /// Answers an ordinary admission the caller is not authorized to make.
    ///
    /// Binding lookup (stages 2-5) runs through the protocol selector; stage-6
    /// connection capacity follows it. `None` means the presenter is
    /// authorized and nothing has been consumed.
    fn classify_record_admission_authority(
        &self,
        request: &RecordAdmission,
        operation_facts: &OperationFacts,
        receiving_epoch: BindingEpoch,
    ) -> Option<ArmOutcome> {
        let binding_detached = BindingState::Detached;
        let (identity, binding) = self.slots.get(&request.participant_id).map_or(
            (
                PresentedIdentity::<Digest, Digest, Digest>::Absent,
                &binding_detached,
            ),
            |slot| {
                (
                    PresentedIdentity::<Digest, Digest, Digest>::Live(&slot.member),
                    &slot.binding,
                )
            },
        );
        if let Some(response) =
            classify_record_admission_binding(identity, binding, receiving_epoch, request)
        {
            return Some(ArmOutcome::respond(response.into_server_value()));
        }
        if let SemanticConnectionCapacityDecision::Respond { limit } =
            operation_facts.semantic_connection_capacity()
        {
            return Some(ArmOutcome::respond(
                RecordAdmissionResponse::connection_conversation_capacity_exceeded(
                    record_envelope(request),
                    limit,
                )
                .into_server_value(),
            ));
        }
        None
    }

    /// Contract amendment A2 (§0.13) defensive idempotence: a committed
    /// identity — the FULL (attempt token, payload fingerprint, verified
    /// participant) triple — re-presented answers ITS commit's exact result
    /// and commits nothing. The triple key is the review-corrected shape
    /// (Cally Ray 2026-08-08, twice): any component demoted from the key to a
    /// checked field leaves one slot per remaining key, so a bypass commit
    /// evicts the demoted identity's answer and an honest answer-lost
    /// re-present of it duplicates — the field bug re-opened through the
    /// eviction. With the whole predicate in the key nothing can evict, a
    /// foreign presenter structurally cannot hit an entry not keyed to it (no
    /// disclosure check needed — the miss IS the guard), and every committed
    /// identity stays independently re-presentable.
    ///
    /// # Contract amendment A4 (§0.15): the same-participant conflict refuses
    ///
    /// A4 converts exactly one of A2's two miss arms. Same token, SAME
    /// verified participant, different canonical payload bytes, committed
    /// match inside the retained op-log window is now a typed
    /// `AttemptTokenBodyConflict::RecordAdmission` refusal that commits
    /// nothing; the other arm — a token hit belonging to a DIFFERENT
    /// participant — keeps A2's warn-and-fall-through exactly, forever.
    ///
    /// ## ⛔ Why this reads two ranges instead of one
    ///
    /// The presenter-scoped range below is a strict SUBSET of the wide range
    /// beneath it, so the natural implementation is to delete one of them and
    /// refuse on the wide hit. That build is outlawed by §0.15 obligation 1:
    /// refusing on a wide hit answers a CROSS-participant token collision, and
    /// any token-correlated answer across participants — refusal, warning,
    /// even latency — is a probe channel that lets one participant test
    /// whether another has spent a token. A4 refuses that leg as a matter of
    /// register law, not as a deferral, and states that a one-widened-arm
    /// implementation "violates the cross-participant clause of this amendment
    /// regardless of its test results". The two ranges are therefore the
    /// mechanism, not a redundancy to be optimised away. The pin standing
    /// under this is
    /// `tests_a4_body_conflict::a_cross_participant_token_hit_still_commits_with_no_refusal`.
    ///
    /// The refusal discloses nothing: the presenter has already been proven by
    /// `classify_record_admission_authority` to BE the committed identity's own
    /// verified participant, so "you already committed different bytes under
    /// this token" reveals only the presenter's own history.
    ///
    /// `None` means no committed identity matched and the caller admits
    /// normally. Consumes nothing on any of the three paths.
    fn answer_committed_record_admission(&self, request: &RecordAdmission) -> Option<ArmOutcome> {
        let dedup_token = request.record_admission_attempt_token.into_bytes();
        let dedup_key = committed_admission_key(request);
        if let Some(committed_delivery_seq) = self.committed_admissions.get(&dedup_key) {
            return Some(ArmOutcome::respond(
                RecordAdmissionResponse::record_committed(RecordCommitted::new(
                    record_envelope(request),
                    *committed_delivery_seq,
                ))
                .into_server_value(),
            ));
        }
        // A4's own range, scoped to the PRESENTER and touching not one foreign
        // entry: the key's participant component sits ahead of the
        // fingerprint precisely so these endpoints select on the presenter
        // (see `committed_admissions`' own doc, where the ordering is the
        // documented mechanism). Reached only after the exact-identity lookup
        // above missed, so any hit here is by construction the same
        // participant's same token under DIFFERENT canonical bytes -- the one
        // conflicting axis this refusal is named for, and the reason it needs
        // no `AttemptConflict` selector.
        if self
            .committed_admissions
            .range(
                (dedup_token, request.participant_id, [0_u8; 32])
                    ..=(dedup_token, request.participant_id, [0xFF_u8; 32]),
            )
            .next()
            .is_some()
        {
            return Some(ArmOutcome::respond(
                RecordAdmissionResponse::attempt_token_body_conflict(
                    request.record_admission_attempt_token,
                    request.conversation_id,
                    request.participant_id,
                    request.capability_generation,
                )
                .into_server_value(),
            ));
        }
        if self
            .committed_admissions
            .range(
                (dedup_token, ParticipantId::MIN, [0_u8; 32])
                    ..=(dedup_token, ParticipantId::MAX, [0xFF_u8; 32]),
            )
            .next()
            .is_some()
        {
            // A2's surviving arm, now reachable ONLY across participants: the
            // presenter-scoped return above has already taken every
            // same-participant hit. The token is committed by SOMEBODY ELSE.
            // Never answered with any prior commit and never refused -- a
            // foreign participant must learn nothing, permanently (A4). Falls
            // through to a normal admission, loudly and only into the server's
            // own log. No sibling delivery sequence is logged: range order is
            // participant order and then fingerprint order, so any single
            // sibling would be an arbitrary one wearing a confident label.
            tracing::warn!(
                conversation_id = self.conversation_id,
                participant_id = request.participant_id,
                "ordinary admission attempt token already committed by a \
                 different participant -- dedup bypassed, committing as a new \
                 record"
            );
        }
        None
    }

    /// Applies one ordinary record admission.
    ///
    /// Binding lookup (stages 2-5), stage-6 connection capacity, and all
    /// frontier-dependent admission outcomes run through protocol selectors.
    /// Commit and mandatory marker-drain arms publish state only after their
    /// complete durable rows have appended and flushed.
    pub(super) fn apply_record_admission_with_impact(
        &mut self,
        request: &RecordAdmission,
        operation_facts: &OperationFacts,
        config: &ParticipantConfig,
        appender: &dyn DurableAppend,
        impact: &mut DispatchImpactAccumulator,
    ) -> Result<ArmOutcome, StateError> {
        let receiving_epoch = BindingEpoch::new(
            operation_facts.receiving_incarnation,
            request.capability_generation,
        );
        if let Some(outcome) =
            self.classify_record_admission_authority(request, operation_facts, receiving_epoch)
        {
            return Ok(outcome);
        }
        // Ordered deliberately AFTER the binding-authority classification
        // above (moving this cheap lookup earlier would hand an unauthorized
        // presenter a token oracle) and BEFORE any frontier or order
        // allocation (a dedup hit consumes no transaction_order major and no
        // delivery sequence).
        if let Some(outcome) = self.answer_committed_record_admission(request) {
            return Ok(outcome);
        }

        let owner = self.take_frontier()?;
        let retained_record_limit = owner.retained_record_limit();
        let (frontiers, closure_accounting, retained_charges, _) = owner.into_parts();
        let slot = self
            .slots
            .get(&request.participant_id)
            .ok_or_else(|| StateError::invariant("authorized record slot disappeared"))?;
        let encoded_record_charge = ordinary_record_charge(request)?;
        let prestate = RecordAdmissionPrestate::new(
            request.clone(),
            PresentedIdentity::<Digest, Digest, Digest>::Live(&slot.member),
            &slot.binding,
            receiving_epoch,
            operation_facts.connection_tracking,
            operation_facts.connection_capacity,
            closure_accounting,
            ResourceVector::new(
                config.max_ordinary_record_entries,
                config.max_ordinary_record_bytes,
            ),
            frontiers,
            retained_charges,
            self.observer_progress,
            ordinary_projection_limits(config),
        );
        match select_record_admission(prestate, encoded_record_charge) {
            RecordAdmissionDecision::Respond(refusal) => {
                let (response, unchanged) = refusal.into_parts();
                let (owner, _, _) = LiveFrontierOwner::from_unchanged_record_admission(
                    unchanged,
                    retained_record_limit,
                );
                self.install_frontier(owner)?;
                Ok(ArmOutcome::respond(response.into_server_value()))
            }
            RecordAdmissionDecision::DrainFirst(drain) => {
                let (candidate, unchanged) = drain.into_parts();
                let (owner, request, _) = LiveFrontierOwner::from_unchanged_record_admission(
                    unchanged,
                    retained_record_limit,
                );
                self.persist_drain_first(candidate, owner, appender, impact)?;
                self.apply_record_admission_with_impact(
                    &request,
                    operation_facts,
                    config,
                    appender,
                    impact,
                )
            }
            // ⛔ TWO DEFECTS WERE ON ONE LINE HERE, and this arm's shape is what
            // keeps them apart. The old body was
            //
            //     let (fault, _) = failure.into_parts();          // owner DROPPED
            //     Err(StateError::invariant(format!(              // #14 funnel
            //         "record admission protocol fault: {fault:?}"
            //     )))
            //
            // The `_` discarded the frontier taken above, leaking the
            // conversation's executable owner; the `StateError::invariant`
            // travelled the #14 funnel (`presented_refusal.rs` module doc) and
            // reached the client as a bare close with NO FRAME, which the
            // 2026-08-28 field carrier read as "fate unknown" and retried
            // forever.
            //
            // `into_terminal_refusal` hands back the response, the fault, and
            // the unchanged aggregate in ONE move, so answering the client and
            // keeping the conversation usable cannot come apart again. Same
            // shape as the Respond arm above; no `PresentedRefusal` carrier is
            // needed, because this IS the arm.
            //
            // ⛔ AND THE TRUTHFUL ANSWER WAS STILL A PERMANENT ONE. The live
            // receipt on a copy of the field estate proved the
            // `MarkerAnchorAccounting` tear RE-FORMS at every carrier attach:
            // attach-time session replay heals four committed rows toward the
            // replay census, the live census then disagrees with the stored
            // ledger, and the next admission faults `{ derived: 1, stored: 0 }`
            // again. A terminal refusal is the truth about that state and the
            // record never lands — and an OLD-WIRE carrier cannot even read the
            // refusal (`0x0127` decodes as `UnknownDiscriminant`, fate UNKNOWN,
            // retry loop). So the class that a census-authoritative reconcile
            // can actually repair gets the repair, and the re-presented
            // admission settles by COMMIT, which every wire vintage can read.
            //
            // Mirror of the replay-side `retry_replay_after_orphan_reconcile`
            // and held to the same discipline: ONE retry, only for this fault
            // class, only when the reconcile actually moved a ledger, and every
            // other outcome falls through to the terminal answer unchanged.
            //
            // The decision is made under `&self` in
            // `answer_record_admission_fault` (the total selector still borrows
            // the participant slot) and only the publication happens here,
            // which is why this arm is two calls and no logic. The fault's full
            // Debug text is logged THERE, in the server's log, where the
            // operator reads it; only the coarse class crosses the wire (R-D1
            // register row, amended 2026-08-28).
            RecordAdmissionDecision::Fault(failure) => {
                match self.answer_record_admission_fault(
                    failure,
                    operation_facts,
                    config,
                    retained_record_limit,
                    receiving_epoch,
                )? {
                    RecordAdmissionFaultAnswer::Committed(commit) => self.persist_record_commit(
                        *commit,
                        receiving_epoch,
                        retained_record_limit,
                        appender,
                        impact,
                    ),
                    RecordAdmissionFaultAnswer::Terminal { response, owner } => {
                        self.install_frontier(*owner)?;
                        Ok(ArmOutcome::respond(response.into_server_value()))
                    }
                }
            }
            RecordAdmissionDecision::Commit(commit) => self.persist_record_commit(
                *commit,
                receiving_epoch,
                retained_record_limit,
                appender,
                impact,
            ),
        }
    }

    /// Decides how one record-admission protocol fault is answered.
    ///
    /// Mints the terminal typed refusal FIRST — the answer to the fault the
    /// selector actually raised — and only then reaches for the one repair the
    /// server has authority to make. Every fault class but
    /// `Projection(MarkerAnchorAccounting)` goes straight to that terminal
    /// answer, exactly as it did at `e7e1da5`.
    ///
    /// Takes `&self`: the returned answer borrows nothing, so the caller's
    /// `&mut self` publication path is reachable immediately after.
    fn answer_record_admission_fault<'a>(
        &'a self,
        failure: Box<RecordAdmissionFailure<'a, Digest, Digest, Digest>>,
        operation_facts: &OperationFacts,
        config: &ParticipantConfig,
        retained_record_limit: u64,
        receiving_epoch: BindingEpoch,
    ) -> Result<RecordAdmissionFaultAnswer, StateError> {
        let census_repairable = matches!(
            failure.fault(),
            RecordAdmissionFault::Projection(
                OrdinaryProjectionError::MarkerAnchorAccounting { .. }
            )
        );
        let (response, fault, unchanged) = failure.into_terminal_refusal();
        let (owner, retry_request, retry_charge) =
            LiveFrontierOwner::from_unchanged_record_admission(unchanged, retained_record_limit);
        let participant_id = retry_request.participant_id;
        let owner = if census_repairable {
            match self.retry_admission_after_anchor_reconcile(
                owner,
                retry_request,
                retry_charge,
                operation_facts,
                config,
                receiving_epoch,
            )? {
                LiveAnchorRetry::Committed(commit) => {
                    return Ok(RecordAdmissionFaultAnswer::Committed(commit));
                }
                LiveAnchorRetry::Unchanged(owner) => *owner,
            }
        } else {
            owner
        };
        // The fault's full `Debug` text stays HERE, in the server's log, where
        // the operator reads it. Only the coarse class crosses the wire (R-D1
        // register row, amended 2026-08-28).
        tracing::error!(
            conversation_id = self.conversation_id,
            participant_id,
            ?fault,
            "record admission protocol fault -- answered with a terminal typed \
             refusal; the conversation keeps its unchanged frontier owner"
        );
        Ok(RecordAdmissionFaultAnswer::Terminal {
            response,
            owner: Box::new(owner),
        })
    }

    /// Reconciles the marker-anchor ledgers of a LIVE admission's unchanged
    /// owner and re-presents the same admission once.
    ///
    /// The live sibling of [`Self::retry_replay_after_orphan_reconcile`], and
    /// deliberately the same shape: reconcile the aggregate the selector handed
    /// back, retry exactly once, and treat a reconcile that MOVED NOTHING as
    /// proof that this fault was never the anchor split. The differences are
    /// the two that matter at a live admission and not at a replay:
    ///
    /// * a replay is auditing a row that already committed, so anything but a
    ///   `Commit` is corruption and raises; here a non-commit is an ordinary
    ///   outcome and simply declines the retry, leaving the caller's terminal
    ///   answer to stand;
    /// * the reconciled owner is RETURNED for installation even when the retry
    ///   does not commit. The census is authoritative on its own evidence and
    ///   does not become wrong because a second, unrelated tear stands beside
    ///   it — and discarding the repair would re-arm this reconcile on every
    ///   subsequent admission, turning one repair into a per-admission loop.
    ///
    /// Takes `&self`: the returned value borrows nothing, so the caller's
    /// `&mut self` publication path is reachable immediately after.
    fn retry_admission_after_anchor_reconcile(
        &self,
        mut owner: LiveFrontierOwner,
        request: RecordAdmission,
        encoded_record_charge: ResourceVector,
        operation_facts: &OperationFacts,
        config: &ParticipantConfig,
        receiving_epoch: BindingEpoch,
    ) -> Result<LiveAnchorRetry, StateError> {
        let retained_record_limit = owner.retained_record_limit();
        // The "reconciled orphaned marker anchors" prefix is the estate's
        // forensics key and stays byte-identical; "at live admission" is this
        // site's suffix, distinct from "at load"
        // (`ops_session_replay.rs::reconcile_load_end_marker_anchors`) and
        // "during replay" (`Self::retry_replay_after_orphan_reconcile`, below
        // in this file), so an operator can tell which of the three repaired
        // the ledger without reading the code.
        let (direction, reconciled) = match owner.reconcile_marker_anchor_ledgers() {
            MarkerAnchorReconcile::RetiredOrphans(orphaned) if orphaned > 0 => {
                ("retired-stored-excess", orphaned)
            }
            MarkerAnchorReconcile::RemintedDeficit(minted) if minted > 0 => {
                ("reminted-census-deficit", minted)
            }
            MarkerAnchorReconcile::RetiredOrphans(_)
            | MarkerAnchorReconcile::RemintedDeficit(_)
            | MarkerAnchorReconcile::InStep
            | MarkerAnchorReconcile::Refused { .. } => {
                return Ok(LiveAnchorRetry::Unchanged(Box::new(owner)));
            }
        };
        tracing::warn!(
            conversation_id = self.conversation_id,
            participant_id = request.participant_id,
            reconciled,
            direction,
            "reconciled orphaned marker anchors at live admission -- the frontier census is \
             authoritative and the refused admission is re-presented once"
        );
        let (frontiers, closure_accounting, retained_charges, _) = owner.into_parts();
        let slot = self
            .slots
            .get(&request.participant_id)
            .ok_or_else(|| StateError::invariant("authorized record slot disappeared"))?;
        let prestate = RecordAdmissionPrestate::new(
            request,
            PresentedIdentity::<Digest, Digest, Digest>::Live(&slot.member),
            &slot.binding,
            receiving_epoch,
            operation_facts.connection_tracking,
            operation_facts.connection_capacity,
            closure_accounting,
            ResourceVector::new(
                config.max_ordinary_record_entries,
                config.max_ordinary_record_bytes,
            ),
            frontiers,
            retained_charges,
            self.observer_progress,
            ordinary_projection_limits(config),
        );
        let unchanged = match select_record_admission(prestate, encoded_record_charge) {
            RecordAdmissionDecision::Commit(commit) => {
                return Ok(LiveAnchorRetry::Committed(commit));
            }
            // A retry that faults AGAIN names a DIFFERENT tear than the one just
            // repaired, and the caller's terminal answer carries only the
            // original fault. Logged here so the operator reads the state that
            // actually stands, not the one that was healed out from under it.
            RecordAdmissionDecision::Fault(failure) => {
                let (retried_fault, unchanged) = failure.into_parts();
                tracing::error!(
                    conversation_id = self.conversation_id,
                    ?retried_fault,
                    "the re-presented admission faulted again after the census reconcile -- \
                     the reconciled ledger is kept and the terminal refusal stands"
                );
                unchanged
            }
            RecordAdmissionDecision::Respond(refusal) => refusal.into_parts().1,
            RecordAdmissionDecision::DrainFirst(drain) => drain.into_parts().1,
        };
        let (owner, _, _) =
            LiveFrontierOwner::from_unchanged_record_admission(unchanged, retained_record_limit);
        Ok(LiveAnchorRetry::Unchanged(Box::new(owner)))
    }

    pub(super) fn persist_next_marker(
        &mut self,
        candidate: ImmutableSequenceCandidate,
        owner: LiveFrontierOwner,
        appender: &dyn DurableAppend,
        impact: &mut DispatchImpactAccumulator,
    ) -> Result<(), StateError> {
        let next_seq =
            candidate
                .delivery_seq()
                .checked_add(1)
                .ok_or(StateError::AllocationExhausted {
                    domain: "delivery sequence after marker drain",
                })?;
        let retained_record_limit = owner.retained_record_limit();
        let marker = canonical_marker_bytes(candidate)?;
        let marker_bytes = u64::try_from(marker.len())
            .map_err(|_| StateError::invariant("canonical marker row length exceeds u64"))?;
        let marker_charge = RetainedRecordCharge::new(
            candidate.delivery_seq(),
            candidate.admission_order(),
            ResourceVector::new(1, marker_bytes),
        );
        let (frontiers, accounting, retained_charges, _) = owner.into_parts();
        let commit = drain_next_marker(frontiers, accounting, retained_charges, marker_charge)
            .map_err(|error| {
                StateError::invariant(format!("mandatory marker drain failed: {error:?}"))
            })?;
        let row = StoredMarkerDrain {
            marker,
            retained_charge: stored_retained_charge(&marker_charge),
            resulting_retained_charges: commit
                .retained_charges()
                .iter()
                .map(stored_retained_charge)
                .collect(),
            successor: format!("{:?}", commit.marker_successor()).into_bytes(),
        };
        let (owner, _, projection) =
            LiveFrontierOwner::from_marker_drain(commit, retained_record_limit);
        validate_marker_projection(self.conversation_id, &projection)?;
        let marker_delivery = projection.delivery().clone();
        #[cfg(test)]
        {
            self.last_marker_projection = Some(marker_delivery.clone());
        }
        let source_log_sequence = self.next_log_sequence;
        let source = StoredOperation::MarkerDrained { row };
        appender.append(&source, source_log_sequence)?;
        self.install_frontier(owner)?;
        self.next_seq = self.next_seq.max(next_seq);
        self.advance_log_head()?;
        self.record_produced_source(
            source_log_sequence,
            &source,
            ReplayedProjectionFacts::marker(marker_delivery),
            appender,
            impact,
        )?;
        if self
            .obligation_debt_dispatch()
            .is_some_and(|state| state.episode().is_some())
        {
            self.record_episode_changed(impact);
        }
        // ⛔ THE CLEARING WRITE (participant contract §0.16 condition 2). This
        // is the `RecordAdmissionDecision::DrainFirst` arm's marker lane and the
        // boot drain's shared core, and it is the ONLY place a `MarkerSettled`
        // wake originates. Recorded AFTER the drain row appended and the
        // resulting owner installed, so the wake cannot promise a candidate is
        // gone before it is.
        //
        // The epoch is `candidate.delivery_seq()`, the head of the immutable
        // sequence lane, which is the same value `precedence_condition`
        // published as `refused_epoch` when it refused this candidate: the
        // stage-11 retry discipline matches the two, so removing either would
        // remove the match.
        impact.record_marker_settled(candidate.delivery_seq());
        Ok(())
    }

    fn persist_record_commit(
        &mut self,
        commit: liminal_protocol::lifecycle::RecordAdmissionCommit,
        receiving_epoch: BindingEpoch,
        retained_record_limit: u64,
        appender: &dyn DurableAppend,
        impact: &mut DispatchImpactAccumulator,
    ) -> Result<ArmOutcome, StateError> {
        let persistence = commit.into_persistence_parts();
        let admission_order = persistence.record.admission_order();
        let connection_capacity = persistence.connection_capacity;
        let row = StoredRecordAdmission {
            request: StoredRecordAdmissionRequest::from(persistence.record.request()),
            receiving_epoch: StoredBindingEpoch::from(receiving_epoch),
            transaction_order: admission_order.transaction_order(),
            delivery_seq: persistence.record.delivery_seq(),
            encoded_record_charge: StoredResourceVector {
                entries: persistence.record.encoded_record_charge().entries,
                bytes: persistence.record.encoded_record_charge().bytes,
            },
            resulting_connection_count: connection_capacity.resulting().occupied(),
            newly_tracked: connection_capacity.newly_tracked(),
            resulting_retained_charges: persistence
                .retained_charges
                .iter()
                .map(stored_retained_charge)
                .collect(),
            resulting_closure_accounting: format!("{:?}", persistence.accounting).into_bytes(),
        };
        let source_log_sequence = self.next_log_sequence;
        let source = StoredOperation::RecordAdmission { row };
        appender.append(&source, source_log_sequence)?;
        let response = persistence.outcome.clone();
        let order = persistence.order.major();
        let sequence = persistence.record.delivery_seq();
        let dedup_key = committed_admission_key(persistence.record.request());
        let owner = LiveFrontierOwner::from_record_admission_persistence(
            persistence,
            retained_record_limit,
        );
        self.install_frontier(owner)?;
        self.committed_admissions.insert(dedup_key, sequence);
        self.observe_replayed_position(order, sequence)?;
        self.advance_log_head()?;
        self.record_produced_source(
            source_log_sequence,
            &source,
            ReplayedProjectionFacts::none(),
            appender,
            impact,
        )?;
        self.record_episode_changed(impact);
        Ok(ArmOutcome::committed(
            RecordAdmissionResponse::record_committed(response).into_server_value(),
            connection_capacity,
        ))
    }

    /// Replays one mandatory v2 marker drain through the protocol-owned drain
    /// and verifies its canonical row, successor, and complete retained charges.
    pub(super) fn replay_marker_drain(
        &mut self,
        row: &StoredMarkerDrain,
    ) -> Result<ParticipantDelivery, StateError> {
        let owner = self.take_frontier()?;
        let retained_record_limit = owner.retained_record_limit();
        let candidate = owner
            .frontiers()
            .sequence()
            .immutable_candidates()
            .first()
            .copied()
            .ok_or_else(|| StateError::invariant("durable marker drain has no candidate"))?;
        let next_seq =
            candidate
                .delivery_seq()
                .checked_add(1)
                .ok_or(StateError::AllocationExhausted {
                    domain: "delivery sequence after durable marker drain",
                })?;
        let marker = canonical_marker_bytes(candidate)?;
        if marker != row.marker {
            return Err(StateError::invariant("durable marker row drifted"));
        }
        let marker_bytes = u64::try_from(marker.len())
            .map_err(|_| StateError::invariant("canonical marker row length exceeds u64"))?;
        let marker_charge = RetainedRecordCharge::new(
            candidate.delivery_seq(),
            candidate.admission_order(),
            ResourceVector::new(1, marker_bytes),
        );
        if stored_retained_charge(&marker_charge) != row.retained_charge {
            return Err(StateError::invariant("durable marker charge drifted"));
        }
        let (frontiers, accounting, retained_charges, _) = owner.into_parts();
        let commit = drain_next_marker(frontiers, accounting, retained_charges, marker_charge)
            .map_err(|error| {
                StateError::invariant(format!("durable marker drain failed: {error:?}"))
            })?;
        let resulting: Vec<_> = commit
            .retained_charges()
            .iter()
            .map(stored_retained_charge)
            .collect();
        if resulting != row.resulting_retained_charges
            || format!("{:?}", commit.marker_successor()).into_bytes() != row.successor
        {
            return Err(StateError::invariant(
                "durable marker drain poststate audit drifted",
            ));
        }
        let (owner, _, projection) =
            LiveFrontierOwner::from_marker_drain(commit, retained_record_limit);
        validate_marker_projection(self.conversation_id, &projection)?;
        self.install_frontier(owner)?;
        self.next_seq = self.next_seq.max(next_seq);
        self.advance_log_head()?;
        Ok(projection.into_delivery())
    }

    /// Re-performs the live load-end orphan reconcile and retries one replayed
    /// admission selection once.
    ///
    /// A committed row is a WITNESS that the live state it committed on held
    /// consistent marker ledgers. The load that served that commit had
    /// reconciled its orphaned anchors at load end — a memory-only repair —
    /// while the replay rebuilt the accounting from rows alone, without it. Any
    /// row committed after that live reconcile therefore re-derives the orphan
    /// split and refuses, making the conversation unloadable (the 2026-08-08
    /// conversation-6 second signature). Re-perform the reconcile at exactly the
    /// first row that proves it happened live, and retry the selection once. A
    /// row that still refuses after an actual retirement falls to the original
    /// invariant unchanged.
    fn retry_replay_after_orphan_reconcile<'a>(
        &'a self,
        failure: Box<RecordAdmissionFailure<'a, Digest, Digest, Digest>>,
        row: &StoredRecordAdmission,
        config: &ParticipantConfig,
        retained_record_limit: u64,
        occupied: u64,
        receiving_epoch: BindingEpoch,
    ) -> Result<Box<RecordAdmissionCommit>, StateError> {
        let (_, unchanged) = failure.into_parts();
        let (mut owner, request, encoded_record_charge) =
            LiveFrontierOwner::from_unchanged_record_admission(unchanged, retained_record_limit);
        // A reconcile that moved nothing keeps the original-invariant
        // fall-through: an in-step ledger (or a repair the closure arithmetic
        // refused) means this row's refusal was never the anchor split, and a
        // fault while replaying a COMMITTED row is still corruption.
        //
        // The "reconciled orphaned marker anchors during replay" prefix below
        // is the estate's forensics key for this repair and stays
        // byte-identical; the direction is a field beside it.
        let (direction, reconciled) = match owner.reconcile_marker_anchor_ledgers() {
            MarkerAnchorReconcile::RetiredOrphans(orphaned) => ("retired-stored-excess", orphaned),
            MarkerAnchorReconcile::RemintedDeficit(minted) => ("reminted-census-deficit", minted),
            MarkerAnchorReconcile::InStep | MarkerAnchorReconcile::Refused { .. } => {
                return Err(StateError::invariant(
                    "durable committed record did not replay as Commit",
                ));
            }
        };
        tracing::warn!(
            conversation_id = self.conversation_id,
            reconciled,
            direction,
            delivery_seq = row.delivery_seq,
            "reconciled orphaned marker anchors during replay -- a committed row \
             witnessed the live load-end reconcile"
        );
        let (frontiers, closure_accounting, retained_charges, _) = owner.into_parts();
        let slot = self
            .slots
            .get(&request.participant_id)
            .ok_or_else(|| StateError::invariant("durable record participant is absent"))?;
        let tracking = if row.newly_tracked {
            ConnectionConversationTracking::Untracked
        } else {
            ConnectionConversationTracking::AlreadyTracked
        };
        let capacity =
            CapacityCounter::try_new(config.max_semantic_conversations_per_connection, occupied)
                .map_err(|error| {
                    StateError::invariant(format!("durable record capacity is invalid: {error:?}"))
                })?;
        let prestate = RecordAdmissionPrestate::new(
            request,
            PresentedIdentity::<Digest, Digest, Digest>::Live(&slot.member),
            &slot.binding,
            receiving_epoch,
            tracking,
            capacity,
            closure_accounting,
            ResourceVector::new(
                config.max_ordinary_record_entries,
                config.max_ordinary_record_bytes,
            ),
            frontiers,
            retained_charges,
            self.observer_progress,
            ordinary_projection_limits(config),
        );
        let RecordAdmissionDecision::Commit(commit) =
            select_record_admission(prestate, encoded_record_charge)
        else {
            return Err(StateError::invariant(
                "durable committed record did not replay as Commit",
            ));
        };
        Ok(commit)
    }

    /// Replays one committed v2 `RecordAdmission` through the same total selector
    /// and verifies every persisted allocation/charge audit before publication.
    pub(super) fn replay_record_admission(
        &mut self,
        row: &StoredRecordAdmission,
        config: &ParticipantConfig,
    ) -> Result<(), StateError> {
        let request = row.request.clone().into_request()?;
        let dedup_key = committed_admission_key(&request);
        let dedup_seq = row.delivery_seq;
        let receiving_epoch = row.receiving_epoch.to_epoch()?;
        let tracking = if row.newly_tracked {
            ConnectionConversationTracking::Untracked
        } else {
            ConnectionConversationTracking::AlreadyTracked
        };
        let occupied = if row.newly_tracked {
            row.resulting_connection_count
                .checked_sub(1)
                .ok_or_else(|| {
                    StateError::invariant(
                        "newly tracked durable record has zero resulting occupancy",
                    )
                })?
        } else {
            row.resulting_connection_count
        };
        let capacity =
            CapacityCounter::try_new(config.max_semantic_conversations_per_connection, occupied)
                .map_err(|error| {
                    StateError::invariant(format!("durable record capacity is invalid: {error:?}"))
                })?;
        let owner = self.take_frontier()?;
        let retained_record_limit = owner.retained_record_limit();
        let (frontiers, closure_accounting, retained_charges, _) = owner.into_parts();
        let slot = self
            .slots
            .get(&request.participant_id)
            .ok_or_else(|| StateError::invariant("durable record participant is absent"))?;
        let encoded_record_charge = ordinary_record_charge(&request)?;
        if encoded_record_charge.entries != row.encoded_record_charge.entries
            || encoded_record_charge.bytes != row.encoded_record_charge.bytes
        {
            return Err(StateError::invariant(
                "durable record canonical charge drifted",
            ));
        }
        let prestate = RecordAdmissionPrestate::new(
            request,
            PresentedIdentity::<Digest, Digest, Digest>::Live(&slot.member),
            &slot.binding,
            receiving_epoch,
            tracking,
            capacity,
            closure_accounting,
            ResourceVector::new(
                config.max_ordinary_record_entries,
                config.max_ordinary_record_bytes,
            ),
            frontiers,
            retained_charges,
            self.observer_progress,
            ordinary_projection_limits(config),
        );
        let commit = match select_record_admission(prestate, encoded_record_charge) {
            RecordAdmissionDecision::Commit(commit) => commit,
            // BOTH SIGNS. The guard was `derived < stored`, matching the
            // one-way reconcile it calls -- so a committed row whose replay
            // re-derived the OTHER split fell straight through to the invariant
            // below and the conversation became unloadable: not a refused
            // record, a refused BOOT. The reconcile is census-authoritative in
            // both directions now, and this guard has to be, or the repair can
            // never be reached from the side the field incident arrives on.
            RecordAdmissionDecision::Fault(failure)
                if matches!(
                    failure.fault(),
                    RecordAdmissionFault::Projection(
                        OrdinaryProjectionError::MarkerAnchorAccounting { .. }
                    )
                ) =>
            {
                self.retry_replay_after_orphan_reconcile(
                    failure,
                    row,
                    config,
                    retained_record_limit,
                    occupied,
                    receiving_epoch,
                )?
            }
            _ => {
                return Err(StateError::invariant(
                    "durable committed record did not replay as Commit",
                ));
            }
        };
        self.publish_replayed_record_admission(
            *commit,
            row,
            retained_record_limit,
            dedup_key,
            dedup_seq,
        )
    }

    /// Verifies every persisted allocation/charge audit of a replayed
    /// admission and only then publishes its state.
    fn publish_replayed_record_admission(
        &mut self,
        commit: RecordAdmissionCommit,
        row: &StoredRecordAdmission,
        retained_record_limit: u64,
        dedup_key: CommittedAdmissionKey,
        dedup_seq: DeliverySeq,
    ) -> Result<(), StateError> {
        let persistence = commit.into_persistence_parts();
        let order = persistence.record.admission_order().transaction_order();
        let sequence = persistence.record.delivery_seq();
        let retained: Vec<_> = persistence
            .retained_charges
            .iter()
            .map(stored_retained_charge)
            .collect();
        if order != row.transaction_order
            || sequence != row.delivery_seq
            || retained != row.resulting_retained_charges
            || persistence.connection_capacity.resulting().occupied()
                != row.resulting_connection_count
            || persistence.connection_capacity.newly_tracked() != row.newly_tracked
            || format!("{:?}", persistence.accounting).into_bytes()
                != row.resulting_closure_accounting
        {
            return Err(StateError::invariant(
                "durable RecordAdmission poststate audit drifted",
            ));
        }
        let owner = LiveFrontierOwner::from_record_admission_persistence(
            persistence,
            retained_record_limit,
        );
        self.install_frontier(owner)?;
        self.committed_admissions.insert(dedup_key, dedup_seq);
        self.observe_replayed_position(order, sequence)?;
        self.advance_log_head()
    }
}

fn validate_marker_projection(
    conversation_id: u64,
    projection: &MarkerDeliveryProjection,
) -> Result<(), StateError> {
    if projection.delivery().conversation_id != conversation_id {
        return Err(StateError::invariant(
            "protocol marker projection belongs to another conversation",
        ));
    }
    Ok(())
}

pub(super) fn canonical_marker_bytes(
    candidate: ImmutableSequenceCandidate,
) -> Result<Vec<u8>, StateError> {
    match candidate {
        ImmutableSequenceCandidate::Marker(marker) => Ok(format!(
            "MarkerCandidateAuthority {{ delivery_seq: {:?}, admission_order: {:?}, target_binding: {:?}, provenance: {:?}, current_owner: {:?} }}",
            marker.delivery_seq,
            marker.admission_order,
            marker.target_binding,
            marker.provenance,
            marker.current_owner,
        )
        .into_bytes()),
        ImmutableSequenceCandidate::BindingTerminal { .. } => Err(StateError::invariant(
            "DrainFirst selected a binding terminal instead of marker work",
        )),
    }
}

const fn stored_retained_charge(
    charge: &liminal_protocol::lifecycle::RetainedRecordCharge,
) -> StoredRetainedCharge {
    let order = charge.admission_order();
    StoredRetainedCharge {
        delivery_seq: charge.delivery_seq(),
        transaction_order: order.transaction_order(),
        candidate_phase: order.candidate_phase() as u8,
        participant_id: order.participant_index(),
        charge: StoredResourceVector {
            entries: charge.encoded_charge().entries,
            bytes: charge.encoded_charge().bytes,
        },
    }
}

/// Builds the echo envelope of one ordinary record admission.
/// Builds one committed ordinary-admission identity key.
///
/// The single construction site of the A2 identity triple, so its component
/// ORDER -- (token, participant, fingerprint), which A4's presenter-scoped
/// range depends on for correctness -- has one place to be right and one place
/// to read about it (`ConversationAuthority::committed_admissions`).
fn committed_admission_key(request: &RecordAdmission) -> CommittedAdmissionKey {
    (
        request.record_admission_attempt_token.into_bytes(),
        request.participant_id,
        ordinary_payload_fingerprint(&request.payload),
    )
}

const fn record_envelope(
    request: &RecordAdmission,
) -> liminal_protocol::wire::RecordAdmissionEnvelope {
    liminal_protocol::wire::RecordAdmissionEnvelope {
        conversation_id: request.conversation_id,
        participant_id: request.participant_id,
        capability_generation: request.capability_generation,
        record_admission_attempt_token: request.record_admission_attempt_token,
    }
}