liminal_protocol/lifecycle/operations/

live_frontier.rs

//! Move-only executable ownership for live lifecycle frontier transitions.
//!
//! Storage supplies canonical encoded row charges, but every participant,
//! binding, cursor, retained row, and aggregate-claim transition is derived
//! inside the protocol from an existing sealed operation commit.

use alloc::{boxed::Box, vec, vec::Vec};
use core::cmp::Ordering;

use crate::{algebra::ResourceVector, wire::RecordAdmission};

use super::super::{
    AttachCommit, AttachTransition, BindingState, ClaimFrontiers, ClosureAccounting, ClosureDebt,
    CommittedDetachTransition, DebtCompletion, DetachCell, DetachedCredentialRecovery, Event,
    FencedAttachCommit, FrontierBinding, FrontierParticipant, IdentityState,
    InitialEnrollmentFrontierCommit, LeaveCommitError, LeaveCommitParameters, LiveMember,
    MarkerAckCommit, NonzeroParticipantAckCommit, ObserverProgressProjection, OrderLedger,
    ParticipantAckCommit, PendingFinalization, PendingLeaveCommitParameters,
    PrepareLeaveAuthorityError, RetainedCausalRecord, RetainedCausalRecordKind, SequenceLedger,
    StoredEdge, VerifiedLeaveRequest,
    claim_frontier::{
        BindingTerminalOwner, FencedMarkerSourceRecord, LiveFrontierTransitionError,
        PrecedenceCondition,
    },
    commit_leave, commit_pending_leave,
};
use super::{
    InitialEnrollmentOperationCommit, MarkerDeliveryProjection, MarkerDrainCommit,
    RecordAdmissionPersistenceParts, RetainedRecordCharge, UnchangedRecordAdmission,
};

mod binding_fate_transition;
mod checkpoint_restore;
pub use checkpoint_restore::{
    CheckpointParticipant, FrontierCheckpointSource, LiveFrontierRestoreError,
    RestoredLiveFrontierOwner,
};
mod ledger;
mod state;
pub(super) use binding_fate_transition::BindingFateOwnerPlan;
use ledger::{
    detach_order, detach_sequence, detached_attach_order, detached_attach_sequence,
    enrollment_order, enrollment_sequence, superseding_attach_order, superseding_attach_sequence,
};
use state::{
    accounting_after_fenced_attach, accounting_after_leave, accounting_after_marker_ack,
    accounting_after_marker_crossings, accounting_after_marker_remint, accounting_after_rows,
    retained_attached, retained_terminal,
};

/// Complete executable frontier, closure-accounting, and keyed-retention owner.
///
/// The owner is intentionally move-only. It is the only live mutation input and
/// restores saved components only against a validated lifecycle capsule.
/// Frontier, closure, retained charges, and participant history therefore cannot
/// be cloned or recombined from different owners:
///
/// ```compile_fail
/// use liminal_protocol::lifecycle::LiveFrontierOwner;
///
/// fn clone_frontier(owner: &LiveFrontierOwner) -> LiveFrontierOwner {
///     owner.clone()
/// }
/// ```
///
/// ```compile_fail
/// use liminal_protocol::lifecycle::LiveFrontierOwner;
///
/// fn splice(left: &mut LiveFrontierOwner, right: LiveFrontierOwner) {
///     left.frontiers = right.frontiers;
///     left.closure_accounting = right.closure_accounting;
///     left.retained_charges = right.retained_charges;
/// }
/// ```
#[derive(Debug, PartialEq, Eq)]
pub struct LiveFrontierOwner {
    frontiers: ClaimFrontiers,
    closure_accounting: ClosureAccounting,
    retained_charges: Vec<RetainedRecordCharge>,
    retained_record_limit: u64,
}

/// Outcome of one marker-anchor ledger reconciliation.
///
/// The two ledgers an admission cross-checks are the frontier CENSUS (marker
/// records whose owning participant is active with a cursor short of the
/// marker, plus markers still pending as immutable sequence candidates) and
/// the STORED count in closure accounting. Reconciliation is
/// census-authoritative: the census is derived from rows that are present, the
/// stored count from arithmetic that a lost row can no longer correct.
///
/// A refused repair is its own arm rather than a zero count. The two are
/// operationally different — one says the ledgers agree, the other says they
/// disagree in a way this repair may not touch — and collapsing them would
/// publish an in-step ledger where a corruption stands.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum MarkerAnchorReconcile {
    /// Census and stored accounting already agree; nothing moved.
    InStep,
    /// Stored accounting held anchors the census can no longer derive; exactly
    /// that excess was retired.
    RetiredOrphans(u64),
    /// The census derives anchors stored accounting has lost; exactly that
    /// deficit was re-minted.
    RemintedDeficit(u64),
    /// The ledgers disagree and `ClosureAccounting::try_new` refused the
    /// repair. State is unchanged, in both directions, exactly as before.
    Refused {
        /// Anchors the frontier census derives.
        census: u64,
        /// Anchors durable closure accounting holds.
        stored: u64,
    },
}

impl MarkerAnchorReconcile {
    /// Returns how many anchors moved, in either direction.
    #[must_use]
    pub const fn reconciled(self) -> u64 {
        match self {
            Self::InStep | Self::Refused { .. } => 0,
            Self::RetiredOrphans(count) | Self::RemintedDeficit(count) => count,
        }
    }
}

impl LiveFrontierOwner {
    pub(super) fn retain_ordinary_cursor_release(
        &mut self,
        source: Box<crate::lifecycle::storage::OrdinaryBindingFateRestore>,
    ) {
        self.frontiers.retain_ordinary_cursor_release(source);
    }

    /// Borrows the accounting scalars; the closure authority remains in the capsule.
    #[must_use]
    pub fn encode_closure_accounting_source(&self) -> Vec<u8> {
        super::super::checkpoint_codec::encode_accounting(&self.closure_accounting)
    }

    /// Decodes accounting against the capsule's validated closure authority.
    ///
    /// # Errors
    ///
    /// Retains framing failures and the native accounting validation error.
    pub fn decode_closure_accounting_source(
        state: super::super::ClosureState,
        bytes: &[u8],
    ) -> Result<ClosureAccounting, super::super::FrontierAccountingDecodeError> {
        super::super::checkpoint_codec::decode_accounting(state, bytes)
    }

    /// Borrows every keyed retained charge in canonical order.
    #[must_use]
    pub fn encode_retained_charge_source(&self) -> Option<Vec<u8>> {
        super::super::checkpoint_codec::encode_charges(&self.retained_charges)
    }

    /// Decodes retained charges without constructing executable frontier authority.
    ///
    /// # Errors
    ///
    /// Refuses malformed framing, non-unit rows and nonconsecutive or unordered keys.
    pub fn decode_retained_charge_source(
        bytes: &[u8],
    ) -> Result<Vec<RetainedRecordCharge>, super::super::CapsuleDecodeError> {
        super::super::checkpoint_codec::decode_charges(bytes)
    }

    /// Acquires live ownership from the protocol's atomic initial-enrollment result.
    #[must_use]
    pub fn from_initial_enrollment<F>(
        initial: InitialEnrollmentFrontierCommit<F>,
        retained_record_limit: u64,
    ) -> (InitialEnrollmentOperationCommit<F>, Self) {
        let (operation, frontiers, closure_accounting, attached_charge) =
            initial.into_conversation_parts();
        let attached = operation.enrollment().attached;
        let retained_charges = vec![RetainedRecordCharge::new(
            attached.delivery_seq(),
            attached.admission_order(),
            attached_charge,
        )];
        (
            operation,
            Self {
                frontiers,
                closure_accounting,
                retained_charges,
                retained_record_limit,
            },
        )
    }

    /// Builds a complete owner from independently validated parts.
    ///
    /// ⛔ ABSENT FROM PRODUCTION BUILDS (`test-support` / `cfg(test)` only) and
    /// deliberately so: it is the one way to express a durable state that NO
    /// live transition can reach, which is exactly what a corruption fixture
    /// needs. The 2026-08-28 marker-anchor wedge — one active participant, one
    /// retained unaccepted compaction marker, stored anchors at zero — cannot
    /// be built any other way, because every live transition moves both marker
    /// ledgers in one commit. Downstream test trees (the server's) reach it
    /// through this seam rather than fabricating a second frontier type.
    #[cfg(any(test, feature = "test-support"))]
    #[must_use]
    pub const fn from_test_parts(
        frontiers: ClaimFrontiers,
        closure_accounting: ClosureAccounting,
        retained_charges: Vec<RetainedRecordCharge>,
        retained_record_limit: u64,
    ) -> Self {
        Self {
            frontiers,
            closure_accounting,
            retained_charges,
            retained_record_limit,
        }
    }

    /// Test-support-only checked capacity extension applied after a real
    /// terminal selector has already chosen Pending. It cannot affect that
    /// disposition; the caller supplies the exact encoded finalizer charges.
    ///
    /// # Errors
    ///
    /// Returns an error if exact capacity arithmetic or reconstruction fails.
    #[cfg(any(test, feature = "test-support"))]
    pub fn with_pending_finalizer_test_capacity(
        mut self,
        finalizer_rows: u64,
        finalizer_charge: ResourceVector,
    ) -> Result<Self, &'static str> {
        let retained_count = u64::try_from(self.retained_charges.len())
            .map_err(|_| "retained record count exceeds u64")?;
        self.retained_record_limit = retained_count
            .checked_add(finalizer_rows)
            .ok_or("pending finalizer retained-record capacity overflow")?;
        let current = self.closure_accounting;
        let configured = current.configured_cap();
        let configured = ResourceVector::new(
            configured
                .entries
                .checked_add(finalizer_charge.entries)
                .ok_or("pending finalizer entry capacity overflow")?,
            configured
                .bytes
                .checked_add(finalizer_charge.bytes)
                .ok_or("pending finalizer byte capacity overflow")?,
        );
        self.closure_accounting = ClosureAccounting::try_new(
            current.state(),
            current.marker_capacity_credits(),
            current.marker_anchors(),
            current.edge_sequence_claims(),
            current.edge_order_position_claims(),
            current.edge_k_remaining(),
            current.baseline(),
            configured,
            current.episode_churn_used(),
            current.episode_churn_limit(),
        )
        .map_err(|_| "pending finalizer closure capacity extension refused")?;
        Ok(self)
    }

    /// Borrows the coupled claim frontiers.
    #[must_use]
    pub const fn frontiers(&self) -> &ClaimFrontiers {
        &self.frontiers
    }

    /// Returns complete current closure accounting.
    #[must_use]
    pub const fn closure_accounting(&self) -> ClosureAccounting {
        self.closure_accounting
    }

    /// Borrows canonical keyed charges for the retained suffix.
    #[must_use]
    pub fn retained_charges(&self) -> &[RetainedRecordCharge] {
        &self.retained_charges
    }

    /// Returns the signed retained causal-row cap.
    #[must_use]
    pub const fn retained_record_limit(&self) -> u64 {
        self.retained_record_limit
    }

    /// Reconciles the stored marker-anchor ledger against the frontier census,
    /// in BOTH directions.
    ///
    /// An anchor is derived only while its marker record survives retention
    /// AND its owning participant is still active with a cursor short of the
    /// marker; participant erasure or record retirement zeroes the derived
    /// side with no acknowledgement row ever retiring the stored side. The
    /// stranded anchor then wedges every later admission on the
    /// `MarkerAnchorAccounting` cross-check with nothing left in the log that
    /// could clear it.
    ///
    /// ⛔ THE OTHER DIRECTION WAS PINNED AS IMPOSSIBLE, AND IT IS NOT. This
    /// repair used to retire the stored excess only, leaving a derived count
    /// at or above the stored one "for the admission projection to fault on".
    /// That was a true statement about LIVE transitions — every one of them
    /// moves both ledgers in a single commit — and a false one about the
    /// system, because this crate's own one-way replay retry mints the
    /// derived-ahead split by over-retiring at a committed row whose replay
    /// census under-derives. The 2026-08-28 field incident is that state
    /// standing on Tom's estate since 2026-08-08 (conversation 1,
    /// `delivery_seq` 187), refusing every admission, forever.
    ///
    /// So the census is the AUTHORITY in both directions: it is derived from
    /// rows that are present, while the stored count is arithmetic a lost row
    /// can no longer correct. `stored > census` retires the excess;
    /// `census > stored` re-mints the deficit. Both are bounded by
    /// `ClosureAccounting::try_new` — a repair the accounting arithmetic
    /// refuses leaves the state unchanged, exactly as the retirement direction
    /// always did.
    ///
    /// The re-mint cannot overshoot. The credit cross-check runs BEFORE the
    /// anchor one and the anchor list is validated as a duplicate-free subset
    /// of the credits (`ordinary_record_projection.rs:983-998` @ `b4ef9fb`), so
    /// any state that reaches the anchor fault has already proven
    /// `derived_anchors <= derived_credits == stored_credits`.
    ///
    /// A PLANNED marker still pending as an immutable sequence candidate holds
    /// its stored anchor before any retained record exists — the `DrainFirst`
    /// discipline keeps admissions (and their cross-check) out of that window.
    /// Those anchors are accounted, not orphaned, and are counted in the census
    /// here the same way.
    #[must_use]
    pub fn reconcile_marker_anchor_ledgers(&mut self) -> MarkerAnchorReconcile {
        let census = self
            .frontiers
            .unaccepted_marker_anchor_count()
            .saturating_add(self.frontiers.pending_marker_candidate_count());
        let stored = self.closure_accounting.marker_anchors();
        let repaired = match census.cmp(&stored) {
            Ordering::Equal => return MarkerAnchorReconcile::InStep,
            Ordering::Less => stored.checked_sub(census).and_then(|orphaned| {
                accounting_after_marker_crossings(self.closure_accounting, orphaned)
                    .map(|accounting| (accounting, MarkerAnchorReconcile::RetiredOrphans(orphaned)))
            }),
            Ordering::Greater => census.checked_sub(stored).and_then(|minted| {
                accounting_after_marker_remint(self.closure_accounting, minted)
                    .map(|accounting| (accounting, MarkerAnchorReconcile::RemintedDeficit(minted)))
            }),
        };
        match repaired {
            Some((accounting, outcome)) => {
                self.closure_accounting = accounting;
                outcome
            }
            None => MarkerAnchorReconcile::Refused { census, stored },
        }
    }

    /// Consumes the complete owner for `RecordAdmission`, Leave, or persistence.
    #[must_use]
    pub fn into_parts(
        self,
    ) -> (
        ClaimFrontiers,
        ClosureAccounting,
        Vec<RetainedRecordCharge>,
        u64,
    ) {
        (
            self.frontiers,
            self.closure_accounting,
            self.retained_charges,
            self.retained_record_limit,
        )
    }

    /// Restores the exact owner returned by a non-committing admission and
    /// recovers the same request for a same-lock retry.
    #[must_use]
    pub fn from_unchanged_record_admission<EF, V, LF>(
        unchanged: UnchangedRecordAdmission<'_, EF, V, LF>,
        retained_record_limit: u64,
    ) -> (Self, RecordAdmission, ResourceVector) {
        let (prestate, encoded_record_charge) = unchanged.into_parts();
        let (request, frontiers, closure_accounting, retained_charges) =
            prestate.into_live_owner_parts();
        (
            Self {
                frontiers,
                closure_accounting,
                retained_charges,
                retained_record_limit,
            },
            request,
            encoded_record_charge,
        )
    }

    /// Acquires the exact owner from the complete sealed successful
    /// `RecordAdmission` persistence authority.
    #[must_use]
    pub fn from_record_admission_persistence(
        persistence: RecordAdmissionPersistenceParts,
        retained_record_limit: u64,
    ) -> Self {
        Self {
            frontiers: persistence.frontiers,
            closure_accounting: persistence.accounting,
            retained_charges: persistence.retained_charges,
            retained_record_limit,
        }
    }

    /// Acquires the exact post-drain owner and durable marker successor.
    #[must_use]
    pub fn from_marker_drain(
        commit: MarkerDrainCommit,
        retained_record_limit: u64,
    ) -> (Self, StoredEdge, MarkerDeliveryProjection) {
        let (frontiers, closure_accounting, retained_charges, successor, projection) =
            commit.into_parts();
        (
            Self {
                frontiers,
                closure_accounting,
                retained_charges,
                retained_record_limit,
            },
            successor,
            projection,
        )
    }

    pub(super) fn commit_binding_terminal_candidate(
        self,
        active_binding: super::super::ActiveBinding,
        admission_order: super::super::AdmissionOrder,
        delivery_seq: crate::wire::DeliverySeq,
        charge: RetainedRecordCharge,
    ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
        let mut active = self.frontiers.active_identities().participants().to_vec();
        let Some(participant) = active
            .iter_mut()
            .find(|participant| participant.participant_index() == active_binding.participant_id)
        else {
            return Err(Box::new((self, LiveFrontierError::Authority)));
        };
        if participant.binding() != FrontierBinding::Bound(active_binding.binding_epoch) {
            return Err(Box::new((self, LiveFrontierError::Authority)));
        }
        *participant = FrontierParticipant::new(
            participant.participant_index(),
            participant.cursor(),
            FrontierBinding::Detached(active_binding.binding_epoch),
        );
        let row = RetainedCausalRecord {
            delivery_seq,
            admission_order,
            kind: RetainedCausalRecordKind::BindingTerminal(super::super::BindingTerminalOwner {
                participant_index: active_binding.participant_id,
                binding_epoch: active_binding.binding_epoch,
            }),
        };
        let Some(sequence) = detach_sequence(self.frontiers.sequence().ledger(), delivery_seq)
        else {
            return Err(Box::new((self, LiveFrontierError::Frontier)));
        };
        let Some(order) = detach_order(
            self.frontiers.order().ledger(),
            admission_order.transaction_order(),
        ) else {
            return Err(Box::new((self, LiveFrontierError::Frontier)));
        };
        match transition(self, (), active, &[row], vec![charge], sequence, order) {
            Ok(committed) => {
                let ((), owner) = committed.into_parts();
                Ok(owner)
            }
            Err(failure) => {
                let error = failure.error();
                let ((), owner) = failure.into_parts();
                Err(Box::new((owner, error)))
            }
        }
    }

    pub(super) fn pend_binding_terminal_candidate(
        self,
        active_binding: super::super::ActiveBinding,
        admission_order: super::super::AdmissionOrder,
        delivery_seq: crate::wire::DeliverySeq,
    ) -> Result<Self, Box<(Self, LiveFrontierError)>> {
        let Some(order) = detach_order(
            self.frontiers.order().ledger(),
            admission_order.transaction_order(),
        ) else {
            return Err(Box::new((self, LiveFrontierError::Frontier)));
        };
        let Self {
            frontiers,
            closure_accounting,
            retained_charges,
            retained_record_limit,
        } = self;
        match frontiers.apply_pending_binding_terminal(
            active_binding.participant_id,
            active_binding.binding_epoch,
            delivery_seq,
            admission_order,
            order,
        ) {
            Ok(frontiers) => Ok(Self {
                frontiers,
                closure_accounting,
                retained_charges,
                retained_record_limit,
            }),
            Err(failure) => {
                let (frontiers, error) = *failure;
                Err(Box::new((
                    Self {
                        frontiers,
                        closure_accounting,
                        retained_charges,
                        retained_record_limit,
                    },
                    map_frontier_error(error),
                )))
            }
        }
    }

    /// Commits the exact first pending binding-terminal candidate as its own
    /// candidate transaction — the R-A2 candidate-lane terminal drain.
    ///
    /// The caller supplies only the pending finalization authority it already
    /// owns and the canonical keyed terminal charge; the candidate itself, its
    /// delivery sequence, and its already-allocated order major are derived
    /// from the coupled frontiers. Retention transitions through the same
    /// accounting the live transitions use, and the resulting owner is exactly
    /// the poststate an immediately-committed terminal would have produced,
    /// with one difference by flavor: a drained `Detached` participant stays
    /// detached at its dead epoch, while a drained `Died` participant leaves
    /// the active identities and frees its identity slot, as a `Left` would
    /// (decision of 11 October 2026, matching the server's erasure of a
    /// drained `Died` slot and token under the 2026-07-23 flavor ruling; see
    /// `ClaimFrontiers::drain_first_binding_terminal`).
    ///
    /// The retained-row CAP is deliberately not re-checked here: the terminal
    /// pended exactly because the cap could not admit its row at fate time
    /// ([`PreparedBindingTerminal::admit`](super::PreparedBindingTerminal::admit)),
    /// and pending DEFERRED that exact reserved row rather than discarding it.
    /// R-A2 prescribes the drain transaction's retention transition
    /// unconditionally — an accepted terminal fate can never become
    /// unsequencable — so the drain honors the deferred reservation even while
    /// the suffix rests at its cap. The overage is bounded by the
    /// open-candidate count, itself bounded by the contract's identity-slot
    /// candidate bound.
    ///
    /// # Errors
    ///
    /// Returns the unchanged owner when the first candidate is absent or not
    /// this pending terminal, the charge does not name the candidate's exact
    /// keyed row, or the resulting retention/closure accounting is invalid.
    pub fn drain_pending_terminal(
        self,
        pending: PendingFinalization,
        terminal_charge: RetainedRecordCharge,
    ) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
        let Some(expected_sequence) = self
            .frontiers
            .sequence()
            .ledger()
            .high_watermark()
            .checked_add(1)
        else {
            return drain_refusal(self, LiveFrontierError::Frontier);
        };
        if terminal_charge.delivery_seq() != expected_sequence
            || terminal_charge.admission_order() != pending.admission_order()
            || terminal_charge.encoded_charge().entries != 1
        {
            return drain_refusal(self, LiveFrontierError::RetainedCharge);
        }
        if self
            .retained_charges
            .len()
            .checked_add(1)
            .and_then(|len| u64::try_from(len).ok())
            .is_none()
        {
            return drain_refusal(self, LiveFrontierError::RetainedRecordLimit);
        }
        let Some(accounting) = accounting_after_rows(self.closure_accounting, &[terminal_charge])
        else {
            return drain_refusal(self, LiveFrontierError::ClosureAccounting);
        };
        let Self {
            frontiers,
            closure_accounting,
            mut retained_charges,
            retained_record_limit,
        } = self;
        let expected_owner = BindingTerminalOwner {
            participant_index: pending.participant_id(),
            binding_epoch: pending.binding_epoch(),
        };
        let (frontiers, record) = match frontiers.drain_first_binding_terminal(
            expected_owner,
            pending.admission_order(),
            matches!(pending, PendingFinalization::Died(_)),
        ) {
            Ok(drained) => drained,
            Err(failure) => {
                let (frontiers, error) = *failure;
                return drain_refusal(
                    Self {
                        frontiers,
                        closure_accounting,
                        retained_charges,
                        retained_record_limit,
                    },
                    map_frontier_error(error),
                );
            }
        };
        retained_charges.push(terminal_charge);
        let projection = ObserverProgressProjection::sourced(
            pending.conversation_id(),
            record.delivery_seq,
            super::super::checkpoint_codec::ObserverProjectionSource::Terminal(
                pending.commit(record.delivery_seq).storage_description(),
            ),
        );
        Ok(DrainedPendingTerminal {
            owner: Self {
                frontiers,
                closure_accounting: accounting,
                retained_charges,
                retained_record_limit,
            },
            projection,
        })
    }

    /// Retains this move-only owner and the exact recovery while its durable
    /// marker source row is read and validated.
    ///
    /// # Errors
    /// Returns the unchanged owner and recovery when the fully restored frontier
    /// does not contain their exact delivered marker occurrence.
    pub fn retain_fenced_marker_source(
        self,
        recovery: DetachedCredentialRecovery,
    ) -> Result<RetainedFencedMarkerSource, Box<FencedMarkerSourceRetentionRefused>> {
        let Some(source) = self.frontiers.fenced_marker_source(recovery) else {
            return Err(Box::new(FencedMarkerSourceRetentionRefused {
                owner: self,
                recovery,
            }));
        };
        Ok(RetainedFencedMarkerSource {
            owner: self,
            recovery,
            expectation: FencedMarkerSourceExpectation { source },
        })
    }

    /// Consumes the complete owner and the exact descriptive inputs to mint one
    /// fenced attach proof from the selected retained marker occurrence.
    ///
    /// This is the sole public production mint. The caller cannot supply a raw
    /// marker token: the owner removes it from its fully validated frontiers.
    /// Refusal returns this owner and every input unchanged after reinstalling
    /// that same occurrence authority, allowing serial retry but no fork.
    #[must_use]
    pub fn mint_fenced_attach(
        mut self,
        marker_source_sequence: u64,
        recovery: DetachedCredentialRecovery,
        debt: ClosureDebt,
        event: Event,
        successor: DebtCompletion,
    ) -> MintFencedAttachResult {
        let Some(record) = self.frontiers.take_fenced_marker_record(recovery) else {
            return MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
                owner: self,
                marker_source_sequence,
                recovery,
                debt,
                event,
                successor,
                reason: FencedAttachMintRefusalReason::MarkerAuthority,
            }));
        };
        let marker_source_bytes = super::super::checkpoint_codec::encode_marker_source(&record);
        match recovery.fenced_attach(record, debt, event, successor) {
            Ok(proof) => MintFencedAttachResult::Minted(Box::new(MintedFencedAttach {
                owner_without_marker_authority: self,
                proof,
                marker_source_sequence,
                marker_source_bytes,
            })),
            Err(refusal) => {
                self.frontiers
                    .reinstall_fenced_marker_record((*refusal).into_record());
                MintFencedAttachResult::MintRefused(Box::new(MintFencedAttachRefused {
                    owner: self,
                    marker_source_sequence,
                    recovery,
                    debt,
                    event,
                    successor,
                    reason: FencedAttachMintRefusalReason::ProofInputs,
                }))
            }
        }
    }
}

/// Complete atomic candidate-lane terminal drain commit.
///
/// The transitioned owner and the protocol-produced observer projection share
/// one sealed predecessor and cannot be recombined from unrelated drains.
#[derive(Debug, PartialEq, Eq)]
pub struct DrainedPendingTerminal {
    owner: LiveFrontierOwner,
    projection: ObserverProgressProjection,
}

impl DrainedPendingTerminal {
    /// Consumes the atomic drain into its owner and exact typed projection.
    #[must_use]
    pub fn into_parts(self) -> (LiveFrontierOwner, ObserverProgressProjection) {
        (self.owner, self.projection)
    }
}

/// Failed candidate-lane terminal drain retaining the unchanged owner.
#[derive(Debug, PartialEq, Eq)]
pub struct PendingTerminalDrainRefused {
    owner: LiveFrontierOwner,
    error: LiveFrontierError,
}

impl PendingTerminalDrainRefused {
    /// Returns the exact typed refusal.
    #[must_use]
    pub const fn error(&self) -> LiveFrontierError {
        self.error
    }

    /// Recovers the unchanged complete owner.
    #[must_use]
    pub fn into_owner(self) -> LiveFrontierOwner {
        self.owner
    }
}

fn drain_refusal(
    owner: LiveFrontierOwner,
    error: LiveFrontierError,
) -> Result<DrainedPendingTerminal, Box<PendingTerminalDrainRefused>> {
    Err(Box::new(PendingTerminalDrainRefused { owner, error }))
}

/// Exact protocol-recomputed marker facts a durable source row must match.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct FencedMarkerSourceExpectation {
    source: FencedMarkerSourceRecord,
}

impl FencedMarkerSourceExpectation {
    /// Returns the conversation owning the durable marker source.
    #[must_use]
    pub const fn conversation_id(self) -> u64 {
        self.source.conversation_id
    }

    /// Returns the marker's durable delivery sequence.
    #[must_use]
    pub const fn marker_delivery_seq(self) -> u64 {
        self.source.delivery_seq
    }

    /// Returns the marker's immutable causal key.
    #[must_use]
    pub const fn admission_order(self) -> super::super::AdmissionOrder {
        self.source.admission_order
    }

    /// Returns the permanent marker owner.
    #[must_use]
    pub const fn participant_id(self) -> u64 {
        self.source.participant_id
    }

    /// Returns the marker's immutable provenance.
    #[must_use]
    pub const fn provenance(self) -> super::super::MarkerProvenance {
        self.source.provenance
    }

    /// Returns the historically validated delivery target.
    #[must_use]
    pub const fn target_binding(self) -> FrontierBinding {
        self.source.target_binding
    }
}

/// Move-only owner/recovery pair held across one bounded durable source read.
#[derive(Debug, PartialEq, Eq)]
pub struct RetainedFencedMarkerSource {
    owner: LiveFrontierOwner,
    recovery: DetachedCredentialRecovery,
    expectation: FencedMarkerSourceExpectation,
}

impl RetainedFencedMarkerSource {
    /// Returns the protocol-recomputed facts the durable row must match.
    #[must_use]
    pub const fn expectation(&self) -> FencedMarkerSourceExpectation {
        self.expectation
    }

    /// Returns the unchanged owner and recovery after source validation.
    #[must_use]
    pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
        (self.owner, self.recovery)
    }
}

/// Refused source retention with the owner and recovery unchanged.
#[derive(Debug, PartialEq, Eq)]
pub struct FencedMarkerSourceRetentionRefused {
    owner: LiveFrontierOwner,
    recovery: DetachedCredentialRecovery,
}

impl FencedMarkerSourceRetentionRefused {
    /// Returns the unchanged owner and recovery after retention refusal.
    #[must_use]
    pub fn into_parts(self) -> (LiveFrontierOwner, DetachedCredentialRecovery) {
        (self.owner, self.recovery)
    }
}

/// Exact reason the owner could not mint a fenced attach proof.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FencedAttachMintRefusalReason {
    /// The validated frontier did not own the selected delivered marker record,
    /// or that record authority was already consumed.
    MarkerAuthority,
    /// Marker authority existed, but recovery/event/successor inputs disagreed.
    ProofInputs,
}

/// Successful one-use fenced proof mint.
#[derive(Debug, PartialEq, Eq)]
pub struct MintedFencedAttach {
    marker_source_sequence: u64,
    marker_source_bytes: Vec<u8>,
    owner_without_marker_authority: LiveFrontierOwner,
    proof: FencedAttachCommit,
}

impl MintedFencedAttach {
    /// Borrows the exact validated marker source before its proof is consumed.
    #[must_use]
    pub fn marker_record_source(&self) -> (u64, &[u8]) {
        (self.marker_source_sequence, &self.marker_source_bytes)
    }

    /// Consumes the result into the owner with spent marker authority and the
    /// sole proof admitted to the downstream by-value chain.
    #[must_use]
    pub fn into_parts(self) -> (LiveFrontierOwner, FencedAttachCommit) {
        (self.owner_without_marker_authority, self.proof)
    }
}

/// Failed one-use fenced proof mint with unchanged retry authority and inputs.
#[derive(Debug, PartialEq, Eq)]
pub struct MintFencedAttachRefused {
    owner: LiveFrontierOwner,
    marker_source_sequence: u64,
    recovery: DetachedCredentialRecovery,
    debt: ClosureDebt,
    event: Event,
    successor: DebtCompletion,
    reason: FencedAttachMintRefusalReason,
}

impl MintFencedAttachRefused {
    /// Returns the typed refusal cause.
    #[must_use]
    pub const fn reason(&self) -> FencedAttachMintRefusalReason {
        self.reason
    }

    /// Consumes the refusal into the unchanged owner and all original inputs.
    #[must_use]
    pub fn into_parts(
        self,
    ) -> (
        LiveFrontierOwner,
        u64,
        DetachedCredentialRecovery,
        ClosureDebt,
        Event,
        DebtCompletion,
    ) {
        (
            self.owner,
            self.marker_source_sequence,
            self.recovery,
            self.debt,
            self.event,
            self.successor,
        )
    }
}

/// Complete result of the sole production fenced proof mint.
#[derive(Debug, PartialEq, Eq)]
pub enum MintFencedAttachResult {
    /// Exactly one marker authority was spent and one proof was minted.
    Minted(Box<MintedFencedAttach>),
    /// No proof was minted; the same authority and inputs are serially retryable.
    MintRefused(Box<MintFencedAttachRefused>),
}

/// Complete move-only settled Leave result: tombstone and executable owner.
#[derive(Debug, PartialEq, Eq)]
pub struct LiveLeaveCommit<EF, V, LF> {
    identity: IdentityState<EF, V, LF>,
    owner: LiveFrontierOwner,
}

impl<EF, V, LF> LiveLeaveCommit<EF, V, LF> {
    /// Projects permanent Leave's exact protocol-committed `Left` sequence.
    #[must_use]
    pub const fn observer_progress_projection(&self) -> Option<ObserverProgressProjection> {
        let IdentityState::Retired(retired) = &self.identity else {
            return None;
        };
        let committed = retired.committed_result();
        Some(ObserverProgressProjection::sourced(
            committed.conversation_id(),
            committed.left_delivery_seq(),
            super::super::checkpoint_codec::ObserverProjectionSource::Leave(
                super::super::LeaveCommittedRestore {
                    conversation_id: committed.conversation_id(),
                    leave_attempt_token: committed.leave_attempt_token(),
                    participant_id: committed.participant_id(),
                    retired_generation: committed.retired_generation(),
                    ended_binding_epoch: committed.ended_binding_epoch(),
                    prior_terminal_delivery_seq: committed.prior_terminal_delivery_seq(),
                    left_delivery_seq: committed.left_delivery_seq(),
                },
            ),
        ))
    }

    /// Consumes the atomic result into its inseparable tombstone and owner.
    #[must_use]
    pub fn into_parts(self) -> (IdentityState<EF, V, LF>, LiveFrontierOwner) {
        (self.identity, self.owner)
    }
}

/// Typed failure of the protocol-owned settled Leave live transition.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LiveLeaveError {
    /// Claim-frontier Leave authority could not be prepared.
    Prepare(PrepareLeaveAuthorityError),
    /// Membership retirement rejected inconsistent authority.
    Commit(LeaveCommitError),
    /// Canonical Left-row charge did not name the protocol-produced row.
    RetainedCharge,
    /// Resulting retained-row count exceeded the signed cap.
    RetainedRecordLimit,
    /// Resulting closure accounting exceeded configured capacity.
    ClosureAccounting,
    /// Leave did not produce a retired identity.
    Identity,
}

/// Commits settled bound or detached Leave through one complete live owner.
///
/// # Errors
///
/// Returns [`LiveLeaveError`] when preparation or retirement authority is
/// inconsistent, the caller's keyed Left charge does not match the committed
/// row, or the resulting retention/closure accounting exceeds its authority.
pub fn commit_settled_leave_frontier<EF, V, LF, D>(
    owner: LiveFrontierOwner,
    member: LiveMember<EF>,
    binding: BindingState,
    detach_cell: DetachCell<D>,
    verified: VerifiedLeaveRequest<V, LF>,
    left_delivery_seq: u64,
    left_charge: RetainedRecordCharge,
) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
    let LiveFrontierOwner {
        frontiers,
        closure_accounting,
        mut retained_charges,
        retained_record_limit,
    } = owner;
    let retired_marker_charge =
        retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
    let authority = frontiers
        .prepare_settled_leave_authority(&member, binding)
        .map_err(LiveLeaveError::Prepare)?;
    let commit = commit_leave(
        member,
        binding,
        detach_cell,
        verified,
        authority,
        LeaveCommitParameters { left_delivery_seq },
    )
    .map_err(LiveLeaveError::Commit)?;
    let (identity, frontiers) = commit.into_parts();
    let IdentityState::Retired(retired) = &identity else {
        return Err(LiveLeaveError::Identity);
    };
    if left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
        || left_charge.admission_order() != retired.left_admission_order()
        || left_charge.encoded_charge().entries != 1
    {
        return Err(LiveLeaveError::RetainedCharge);
    }
    retained_charges.push(left_charge);
    retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
    let retained_len = u64::try_from(frontiers.retained_records().len())
        .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
    if retained_len > retained_record_limit
        || retained_charges.len() != frontiers.retained_records().len()
    {
        return Err(LiveLeaveError::RetainedRecordLimit);
    }
    let closure_accounting =
        accounting_after_leave(closure_accounting, &[left_charge], retired_marker_charge)
            .ok_or(LiveLeaveError::ClosureAccounting)?;
    Ok(LiveLeaveCommit {
        identity,
        owner: LiveFrontierOwner {
            frontiers,
            closure_accounting,
            retained_charges,
            retained_record_limit,
        },
    })
}

/// Commits a pending binding terminal immediately before Leave through one
/// complete live owner.
///
/// # Errors
///
/// Returns [`LiveLeaveError`] when pending preparation or retirement authority
/// is inconsistent, either caller charge does not match its protocol-produced
/// row, or resulting retention/closure accounting exceeds its authority.
pub fn commit_pending_leave_frontier<EF, V, LF, D>(
    owner: LiveFrontierOwner,
    member: LiveMember<EF>,
    pending: PendingFinalization,
    detach_cell: DetachCell<D>,
    verified: VerifiedLeaveRequest<V, LF>,
    parameters: PendingLeaveCommitParameters,
    charges: [RetainedRecordCharge; 2],
) -> Result<LiveLeaveCommit<EF, V, LF>, LiveLeaveError> {
    let [terminal_charge, left_charge] = charges;
    let terminal_delivery_seq = parameters.terminal_delivery_seq;
    let LiveFrontierOwner {
        frontiers,
        closure_accounting,
        mut retained_charges,
        retained_record_limit,
    } = owner;
    let retired_marker_charge =
        retired_marker_charge(&frontiers, &retained_charges, member.participant_id())?;
    let authority = frontiers
        .prepare_pending_leave_authority(&member, pending)
        .map_err(LiveLeaveError::Prepare)?;
    let commit = commit_pending_leave(
        member,
        pending,
        detach_cell,
        verified,
        authority,
        parameters,
    )
    .map_err(LiveLeaveError::Commit)?;
    let (identity, frontiers) = commit.into_parts();
    let IdentityState::Retired(retired) = &identity else {
        return Err(LiveLeaveError::Identity);
    };
    if retired.committed_result().prior_terminal_delivery_seq() != Some(terminal_delivery_seq)
        || terminal_charge.delivery_seq() != terminal_delivery_seq
        || terminal_charge.admission_order() != pending.admission_order()
        || terminal_charge.encoded_charge().entries != 1
        || left_charge.delivery_seq() != retired.committed_result().left_delivery_seq()
        || left_charge.admission_order() != retired.left_admission_order()
        || left_charge.encoded_charge().entries != 1
    {
        return Err(LiveLeaveError::RetainedCharge);
    }
    retained_charges.extend([terminal_charge, left_charge]);
    retained_charges.sort_unstable_by_key(|charge| charge.delivery_seq());
    let retained_len = u64::try_from(frontiers.retained_records().len())
        .map_err(|_| LiveLeaveError::RetainedRecordLimit)?;
    if retained_len > retained_record_limit {
        return Err(LiveLeaveError::RetainedRecordLimit);
    }
    if retained_charges.len() != frontiers.retained_records().len() {
        return Err(LiveLeaveError::RetainedCharge);
    }
    let closure_accounting = accounting_after_leave(
        closure_accounting,
        &[terminal_charge, left_charge],
        retired_marker_charge,
    )
    .ok_or(LiveLeaveError::ClosureAccounting)?;
    Ok(LiveLeaveCommit {
        identity,
        owner: LiveFrontierOwner {
            frontiers,
            closure_accounting,
            retained_charges,
            retained_record_limit,
        },
    })
}

fn retired_marker_charge(
    frontiers: &ClaimFrontiers,
    retained_charges: &[RetainedRecordCharge],
    participant_id: crate::wire::ParticipantId,
) -> Result<Option<RetainedRecordCharge>, LiveLeaveError> {
    let marker_sequence = frontiers
        .retained_marker_records()
        .iter()
        .find_map(|record| {
            matches!(
                record.kind,
                RetainedCausalRecordKind::CompactionMarker {
                    participant_index,
                    ..
                } if participant_index == participant_id
            )
            .then_some(record.delivery_seq)
        });
    let Some(marker_sequence) = marker_sequence else {
        return Ok(None);
    };
    retained_charges
        .iter()
        .copied()
        .find(|charge| charge.delivery_seq() == marker_sequence)
        .map(Some)
        .ok_or(LiveLeaveError::RetainedCharge)
}

/// Exact charges for a credential attach's one or two retained rows.
#[derive(Debug, PartialEq, Eq)]
pub struct AttachFrontierCharges {
    terminal: Option<RetainedRecordCharge>,
    attached: RetainedRecordCharge,
    seal: LiveTransitionInputSeal,
}

#[derive(Debug, PartialEq, Eq)]
enum LiveTransitionInputSeal {
    Validated,
}

impl AttachFrontierCharges {
    /// Couples the canonical `Attached` charge with an optional terminal charge.
    #[must_use]
    pub const fn new(
        terminal: Option<RetainedRecordCharge>,
        attached: RetainedRecordCharge,
    ) -> Self {
        Self {
            terminal,
            attached,
            seal: LiveTransitionInputSeal::Validated,
        }
    }

    const fn into_parts(self) -> (Option<RetainedRecordCharge>, RetainedRecordCharge) {
        let Self {
            terminal,
            attached,
            seal,
        } = self;
        match seal {
            LiveTransitionInputSeal::Validated => (terminal, attached),
        }
    }
}

/// A typed lifecycle commit paired with its complete post-transition owner.
#[derive(Debug, PartialEq, Eq)]
pub struct LiveFrontierCommit<T> {
    operation: T,
    owner: LiveFrontierOwner,
}

impl<T> LiveFrontierCommit<T> {
    /// Borrows the exact typed lifecycle commit.
    #[must_use]
    pub const fn operation(&self) -> &T {
        &self.operation
    }

    /// Borrows the complete post-transition owner.
    #[must_use]
    pub const fn owner(&self) -> &LiveFrontierOwner {
        &self.owner
    }

    /// Consumes the atomic transition for durability publication.
    #[must_use]
    pub fn into_parts(self) -> (T, LiveFrontierOwner) {
        (self.operation, self.owner)
    }
}

/// Failed live transition retaining the unchanged complete owner and operation.
#[derive(Debug, PartialEq, Eq)]
pub struct LiveFrontierFailure<T> {
    error: LiveFrontierError,
    operation: T,
    owner: LiveFrontierOwner,
}

impl<T> LiveFrontierFailure<T> {
    /// Returns the exact typed transition failure.
    #[must_use]
    pub const fn error(&self) -> LiveFrontierError {
        self.error
    }

    /// Recovers the unchanged owner and intact operation commit.
    #[must_use]
    pub fn into_parts(self) -> (T, LiveFrontierOwner) {
        (self.operation, self.owner)
    }
}

/// Failure selected while coupling a sealed lifecycle commit to live ownership.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LiveFrontierError {
    /// Commit and live owner name different authority.
    Authority,
    /// A mandatory immutable/recovery transition has precedence.
    ///
    /// Carries WHICH of amendment A5's clearing conditions blocked
    /// (participant contract §0.16). The law attaches to the SEAM, so the
    /// discriminant is produced by `apply_live_transition`'s own guard and
    /// carried out through every wrapper — a future wrapper inherits it
    /// without being enumerated anywhere.
    Precedence(PrecedenceCondition),
    /// Canonical keyed row charges differ from the commit-derived retained rows.
    RetainedCharge,
    /// The retained causal-row cap would be exceeded.
    RetainedRecordLimit,
    /// Aggregate claim arithmetic or exact owner reconstruction failed.
    Frontier,
    /// Resulting closure accounting is invalid or outside its signed capacity.
    ClosureAccounting,
}

/// Result of coupling any typed lifecycle commit to live frontier ownership.
pub type LiveFrontierResult<T> = Result<LiveFrontierCommit<T>, Box<LiveFrontierFailure<T>>>;

/// Applies a subsequent enrollment to the complete live owner.
///
/// # Errors
///
/// Returns a failure retaining the unchanged owner and intact enrollment commit.
pub fn apply_enrollment_frontier<F>(
    owner: LiveFrontierOwner,
    operation: super::super::EnrollmentCommit<F>,
    attached_charge: RetainedRecordCharge,
) -> LiveFrontierResult<super::super::EnrollmentCommit<F>> {
    let attached = operation.attached;
    if attached.conversation_id() != owner.frontiers.conversation_id() {
        return failure(owner, operation, LiveFrontierError::Authority);
    }
    let participant_id = attached.participant_id();
    let mut active = owner.frontiers.active_identities().participants().to_vec();
    if active
        .iter()
        .any(|participant| participant.participant_index() == participant_id)
    {
        return failure(owner, operation, LiveFrontierError::Authority);
    }
    active.push(FrontierParticipant::new(
        participant_id,
        operation.member.cursor(),
        FrontierBinding::Bound(attached.binding_epoch()),
    ));
    active.sort_unstable_by_key(|participant| participant.participant_index());
    let rows = [retained_attached(attached)];
    let Some(sequence) =
        enrollment_sequence(owner.frontiers.sequence().ledger(), attached.delivery_seq())
    else {
        return failure(owner, operation, LiveFrontierError::Frontier);
    };
    let Some(order) = enrollment_order(
        owner.frontiers.order().ledger(),
        attached.admission_order().transaction_order(),
    ) else {
        return failure(owner, operation, LiveFrontierError::Frontier);
    };
    transition(
        owner,
        operation,
        active,
        &rows,
        vec![attached_charge],
        sequence,
        order,
    )
}

/// Applies credential attach to the complete live owner.
///
/// # Errors
///
/// Returns a failure retaining the unchanged owner, intact attach commit, and
/// exact reason the commit could not enter the frontier.
pub fn apply_attach_frontier<F, V>(
    owner: LiveFrontierOwner,
    operation: AttachCommit<F, V>,
    charges: AttachFrontierCharges,
) -> LiveFrontierResult<AttachCommit<F, V>> {
    let (terminal_charge, attached_charge) = charges.into_parts();
    let attached = operation.attached;
    if attached.conversation_id() != owner.frontiers.conversation_id() {
        return failure(owner, operation, LiveFrontierError::Authority);
    }
    let mut active = owner.frontiers.active_identities().participants().to_vec();
    let Some(participant) = active
        .iter_mut()
        .find(|participant| participant.participant_index() == attached.participant_id())
    else {
        return failure(owner, operation, LiveFrontierError::Authority);
    };
    let frontier_cursor = participant.cursor();
    *participant = FrontierParticipant::new(
        participant.participant_index(),
        operation.member.cursor(),
        FrontierBinding::Bound(attached.binding_epoch()),
    );
    let reopened = reopened_marker_anchors(
        &owner,
        attached.participant_id(),
        frontier_cursor,
        operation.member.cursor(),
    );
    with_reopened_marker_anchors(owner, operation, reopened, |owner, operation| {
        apply_attach_rows(owner, operation, active, terminal_charge, attached_charge)
    })
}

/// Commits a detached or superseding attach's rows once its cursor is placed.
fn apply_attach_rows<F, V>(
    owner: LiveFrontierOwner,
    operation: AttachCommit<F, V>,
    active: Vec<FrontierParticipant>,
    terminal_charge: Option<RetainedRecordCharge>,
    attached_charge: RetainedRecordCharge,
) -> LiveFrontierResult<AttachCommit<F, V>> {
    let attached = operation.attached;
    let current_sequence = owner.frontiers.sequence().ledger();
    let current_order = owner.frontiers.order().ledger();
    let (rows, keyed_charges, sequence, order) = match operation.transition {
        AttachTransition::Detached => {
            if terminal_charge.is_some() {
                return failure(owner, operation, LiveFrontierError::RetainedCharge);
            }
            let Some(sequence) =
                detached_attach_sequence(current_sequence, attached.delivery_seq())
            else {
                return failure(owner, operation, LiveFrontierError::Frontier);
            };
            let Some(order) = detached_attach_order(
                current_order,
                attached.admission_order().transaction_order(),
            ) else {
                return failure(owner, operation, LiveFrontierError::Frontier);
            };
            (
                vec![retained_attached(attached)],
                vec![attached_charge],
                sequence,
                order,
            )
        }
        AttachTransition::Superseded { terminal } => {
            let Some(terminal_charge) = terminal_charge else {
                return failure(owner, operation, LiveFrontierError::RetainedCharge);
            };
            let rows = vec![
                retained_terminal(terminal.into()),
                retained_attached(attached),
            ];
            let Some(sequence) = superseding_attach_sequence(current_sequence, &rows) else {
                return failure(owner, operation, LiveFrontierError::Frontier);
            };
            let Some(order) = superseding_attach_order(
                current_order,
                attached.admission_order().transaction_order(),
            ) else {
                return failure(owner, operation, LiveFrontierError::Frontier);
            };
            (
                rows,
                vec![terminal_charge, attached_charge],
                sequence,
                order,
            )
        }
        AttachTransition::FencedRecovery {
            prior_binding_epoch,
            composed_terminal,
            next_closure_state,
        } => {
            return apply_fenced_attach_frontier(
                owner,
                operation,
                terminal_charge,
                attached_charge,
                prior_binding_epoch,
                composed_terminal,
                next_closure_state,
            );
        }
    };
    transition(
        owner,
        operation,
        active,
        &rows,
        keyed_charges,
        sequence,
        order,
    )
}

/// Applies a committed detach terminal to the complete live owner.
///
/// # Errors
///
/// Returns a failure retaining the unchanged owner and intact detach commit.
pub fn apply_detach_frontier<EF, V>(
    owner: LiveFrontierOwner,
    operation: CommittedDetachTransition<EF, V>,
    terminal_charge: RetainedRecordCharge,
) -> LiveFrontierResult<CommittedDetachTransition<EF, V>> {
    let terminal = operation.terminal();
    if terminal.conversation_id() != owner.frontiers.conversation_id() {
        return failure(owner, operation, LiveFrontierError::Authority);
    }
    let mut active = owner.frontiers.active_identities().participants().to_vec();
    let Some(participant) = active
        .iter_mut()
        .find(|participant| participant.participant_index() == terminal.participant_id())
    else {
        return failure(owner, operation, LiveFrontierError::Authority);
    };
    *participant = FrontierParticipant::new(
        participant.participant_index(),
        operation.member().cursor(),
        FrontierBinding::Detached(terminal.binding_epoch()),
    );
    let row = retained_terminal(terminal.into());
    let Some(sequence) = detach_sequence(owner.frontiers.sequence().ledger(), row.delivery_seq)
    else {
        return failure(owner, operation, LiveFrontierError::Frontier);
    };
    let Some(order) = detach_order(
        owner.frontiers.order().ledger(),
        row.admission_order.transaction_order(),
    ) else {
        return failure(owner, operation, LiveFrontierError::Frontier);
    };
    transition(
        owner,
        operation,
        active,
        &[row],
        vec![terminal_charge],
        sequence,
        order,
    )
}

/// Counts this participant's active marker anchors that a cumulative cursor
/// advance from `previous_cursor` to `through_seq` crosses.
///
/// A crossed anchor is one that was unaccepted before the advance
/// (`previous_cursor < marker seq`) and is accepted after it
/// (`marker seq <= through_seq`) — the exact transition the admission
/// projection's derived side (`ordinary_unaccepted_marker_anchors`) observes
/// from the cursor, which is why the stored accounting must move by the same
/// count in the same commit.
fn crossed_marker_anchors(
    owner: &LiveFrontierOwner,
    participant_id: crate::wire::ParticipantId,
    previous_cursor: crate::wire::DeliverySeq,
    through_seq: crate::wire::DeliverySeq,
) -> u64 {
    owner
        .frontiers
        .retained_marker_records()
        .iter()
        .filter(|record| {
            matches!(
                record.kind,
                RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
                    if participant_index == participant_id
            ) && previous_cursor < record.delivery_seq
                && record.delivery_seq <= through_seq
        })
        .fold(0_u64, |count, _| count.saturating_add(1))
}

/// Applies a zero-debt participant acknowledgement cursor transition.
///
/// A cumulative ack whose `through_seq` crosses this participant's delivered
/// compaction marker accepts it, so the marker's anchor is retired from the
/// closure accounting in the same commit — the ordinary-path sibling of
/// [`apply_marker_ack_frontier`]. Left unretired, the stored anchor count and
/// the cursor-derived one diverge and every later admission faults
/// (`MarkerAnchorAccounting`), which the dispatch funnel presents as a silent
/// connection close (the 2026-08-07 manifold wedge).
///
/// # Errors
///
/// Returns a failure retaining the unchanged owner and intact ack commit.
pub fn apply_participant_ack_frontier(
    mut owner: LiveFrontierOwner,
    operation: ParticipantAckCommit,
) -> LiveFrontierResult<ParticipantAckCommit> {
    let request = operation.outcome().request();
    let Some(current) = owner
        .frontiers
        .active_identities()
        .participants()
        .iter()
        .find(|participant| participant.participant_index() == request.participant_id)
        .copied()
    else {
        return failure(owner, operation, LiveFrontierError::Authority);
    };
    let crossed = crossed_marker_anchors(
        &owner,
        request.participant_id,
        current.cursor(),
        request.through_seq,
    );
    let accounting = if crossed == 0 {
        None
    } else {
        match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
            Some(accounting) => Some(accounting),
            None => {
                return failure(owner, operation, LiveFrontierError::ClosureAccounting);
            }
        }
    };
    let participant = FrontierParticipant::new(
        request.participant_id,
        request.through_seq,
        current.binding(),
    );
    owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
        Ok(frontiers) => frontiers,
        Err(frontier_failure) => {
            let (frontiers, error) = *frontier_failure;
            owner.frontiers = frontiers;
            return failure(owner, operation, map_frontier_error(error));
        }
    };
    if let Some(accounting) = accounting {
        owner.closure_accounting = accounting;
    }
    Ok(LiveFrontierCommit { operation, owner })
}

/// Applies a nonzero-debt participant acknowledgement cursor transition.
///
/// The episode and member remain owned by the sealed aggregate commit; this
/// transition consumes the same exact acknowledged cursor into the coupled
/// claim-frontier participant rank.
///
/// # Errors
///
/// Returns a failure retaining the unchanged owner and intact aggregate commit.
pub fn apply_nonzero_participant_ack_frontier(
    mut owner: LiveFrontierOwner,
    operation: NonzeroParticipantAckCommit,
) -> LiveFrontierResult<NonzeroParticipantAckCommit> {
    let request = operation.outcome().request();
    let Some(current) = owner
        .frontiers
        .active_identities()
        .participants()
        .iter()
        .find(|participant| participant.participant_index() == request.participant_id)
        .copied()
    else {
        return failure(owner, operation, LiveFrontierError::Authority);
    };
    // A nonzero-debt cumulative ack crosses markers exactly as the zero-debt
    // one does; the anchor accounting moves with the cursor here too.
    let crossed = crossed_marker_anchors(
        &owner,
        request.participant_id,
        current.cursor(),
        request.through_seq,
    );
    let accounting = if crossed == 0 {
        None
    } else {
        match accounting_after_marker_crossings(owner.closure_accounting, crossed) {
            Some(accounting) => Some(accounting),
            None => {
                return failure(owner, operation, LiveFrontierError::ClosureAccounting);
            }
        }
    };
    let participant = FrontierParticipant::new(
        request.participant_id,
        request.through_seq,
        current.binding(),
    );
    owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
        Ok(frontiers) => frontiers,
        Err(frontier_failure) => {
            let (frontiers, error) = *frontier_failure;
            owner.frontiers = frontiers;
            return failure(owner, operation, map_frontier_error(error));
        }
    };
    if let Some(accounting) = accounting {
        owner.closure_accounting = accounting;
    }
    Ok(LiveFrontierCommit { operation, owner })
}

/// Applies a zero-debt marker acknowledgement cursor transition.
///
/// # Errors
///
/// Returns a failure retaining the unchanged owner and intact marker-ack commit.
pub fn apply_marker_ack_frontier(
    mut owner: LiveFrontierOwner,
    operation: MarkerAckCommit,
) -> LiveFrontierResult<MarkerAckCommit> {
    let request = operation.outcome().request();
    if !owner
        .frontiers
        .retained_marker_records()
        .iter()
        .any(|record| {
            record.delivery_seq == request.marker_delivery_seq
                && matches!(
                    record.kind,
                    RetainedCausalRecordKind::CompactionMarker { participant_index, .. }
                        if participant_index == request.participant_id
                )
        })
    {
        return failure(owner, operation, LiveFrontierError::Authority);
    }
    let Some(current) = owner
        .frontiers
        .active_identities()
        .participants()
        .iter()
        .find(|participant| participant.participant_index() == request.participant_id)
        .copied()
    else {
        return failure(owner, operation, LiveFrontierError::Authority);
    };
    let Some(accounting) = accounting_after_marker_ack(owner.closure_accounting) else {
        return failure(owner, operation, LiveFrontierError::ClosureAccounting);
    };
    let participant = FrontierParticipant::new(
        request.participant_id,
        request.marker_delivery_seq,
        current.binding(),
    );
    owner.frontiers = match owner.frontiers.apply_live_identity(participant) {
        Ok(frontiers) => frontiers,
        Err(frontier_failure) => {
            let (frontiers, error) = *frontier_failure;
            owner.frontiers = frontiers;
            return failure(owner, operation, map_frontier_error(error));
        }
    };
    owner.closure_accounting = accounting;
    Ok(LiveFrontierCommit { operation, owner })
}

fn apply_fenced_attach_frontier<F, V>(
    owner: LiveFrontierOwner,
    operation: AttachCommit<F, V>,
    terminal_charge: Option<RetainedRecordCharge>,
    attached_charge: RetainedRecordCharge,
    prior_binding_epoch: crate::wire::BindingEpoch,
    composed_terminal: Option<super::super::CommittedBindingTerminal>,
    next_closure_state: super::super::ClosureState,
) -> LiveFrontierResult<AttachCommit<F, V>> {
    let attached = operation.attached;
    let (rows, charges) = match (composed_terminal, terminal_charge) {
        (None, None) => (vec![retained_attached(attached)], vec![attached_charge]),
        (Some(terminal), Some(terminal_charge)) => (
            vec![retained_terminal(terminal), retained_attached(attached)],
            vec![terminal_charge, attached_charge],
        ),
        (None, Some(_)) | (Some(_), None) => {
            return failure(owner, operation, LiveFrontierError::RetainedCharge);
        }
    };
    let participant = FrontierParticipant::new(
        attached.participant_id(),
        operation.member.cursor(),
        FrontierBinding::Bound(attached.binding_epoch()),
    );
    fenced_attach_transition(
        owner,
        operation,
        participant,
        prior_binding_epoch,
        next_closure_state,
        &rows,
        charges,
    )
}

/// Counts the marker anchors an attach re-opens by placing its participant's
/// cursor below where the frontier held it.
///
/// An attach installs the MEMBER's cursor, the participant's durable
/// acknowledged boundary. A connection fate does not move that boundary: it
/// carries only the frontier's copy of the cursor to the high watermark
/// (`ClaimFrontiers::prepare_binding_fate_transition`), and retires the anchors
/// of every marker that carry passes. When the same participant attaches
/// again, its cursor returns to the boundary it really holds, and every
/// marker of its own between that boundary and the carried cursor is
/// unaccepted again — the census derives its anchor once more. The stored
/// ledger has to re-mint exactly those anchors in the same commit, or the
/// census runs one ahead of it and every boot after re-mints the deficit at
/// load, the mirror image of the orphans the fate used to leave.
///
/// Zero whenever the attach does not lower the cursor, which is every attach
/// that no non-finalizer fate preceded.
fn reopened_marker_anchors(
    owner: &LiveFrontierOwner,
    participant_id: crate::wire::ParticipantId,
    frontier_cursor: crate::wire::DeliverySeq,
    member_cursor: crate::wire::DeliverySeq,
) -> u64 {
    if member_cursor >= frontier_cursor {
        return 0;
    }
    crossed_marker_anchors(owner, participant_id, member_cursor, frontier_cursor)
}

/// Runs an attach transition over closure accounting that holds the anchors it
/// re-opens, and hands back the unchanged owner if the transition refuses.
fn with_reopened_marker_anchors<T>(
    mut owner: LiveFrontierOwner,
    operation: T,
    reopened: u64,
    apply: impl FnOnce(LiveFrontierOwner, T) -> LiveFrontierResult<T>,
) -> LiveFrontierResult<T> {
    if reopened == 0 {
        return apply(owner, operation);
    }
    let unchanged = owner.closure_accounting;
    let Some(accounting) = accounting_after_marker_remint(unchanged, reopened) else {
        return failure(owner, operation, LiveFrontierError::ClosureAccounting);
    };
    owner.closure_accounting = accounting;
    apply(owner, operation).map_err(|mut refused| {
        refused.owner.closure_accounting = unchanged;
        refused
    })
}

fn fenced_attach_transition<T>(
    mut owner: LiveFrontierOwner,
    operation: T,
    participant: FrontierParticipant,
    prior_binding_epoch: crate::wire::BindingEpoch,
    next_closure_state: super::super::ClosureState,
    rows: &[RetainedCausalRecord],
    charges: Vec<RetainedRecordCharge>,
) -> LiveFrontierResult<T> {
    if rows.len() != charges.len()
        || rows.iter().zip(&charges).any(|(row, charge)| {
            row.delivery_seq != charge.delivery_seq()
                || row.admission_order != charge.admission_order()
                || charge.encoded_charge().entries != 1
        })
    {
        return failure(owner, operation, LiveFrontierError::RetainedCharge);
    }
    let resulting_len = owner
        .frontiers
        .retained_records()
        .len()
        .checked_add(rows.len());
    if resulting_len
        .and_then(|len| u64::try_from(len).ok())
        .is_none_or(|len| len > owner.retained_record_limit)
    {
        return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
    }
    let Some(accounting) =
        accounting_after_fenced_attach(owner.closure_accounting, &charges, next_closure_state)
    else {
        return failure(owner, operation, LiveFrontierError::ClosureAccounting);
    };
    owner.frontiers =
        match owner
            .frontiers
            .apply_live_fenced_attach(participant, prior_binding_epoch, rows)
        {
            Ok(frontiers) => frontiers,
            Err(frontier_failure) => {
                let (frontiers, error) = *frontier_failure;
                owner.frontiers = frontiers;
                return failure(owner, operation, map_frontier_error(error));
            }
        };
    owner.retained_charges.extend(charges);
    owner
        .retained_charges
        .sort_unstable_by_key(|charge| charge.delivery_seq());
    owner.closure_accounting = accounting;
    Ok(LiveFrontierCommit { operation, owner })
}

fn transition<T>(
    mut owner: LiveFrontierOwner,
    operation: T,
    active: Vec<FrontierParticipant>,
    rows: &[RetainedCausalRecord],
    charges: Vec<RetainedRecordCharge>,
    sequence: SequenceLedger,
    order: OrderLedger,
) -> LiveFrontierResult<T> {
    if rows.len() != charges.len()
        || rows.iter().zip(&charges).any(|(row, charge)| {
            row.delivery_seq != charge.delivery_seq()
                || row.admission_order != charge.admission_order()
                || charge.encoded_charge().entries != 1
        })
    {
        return failure(owner, operation, LiveFrontierError::RetainedCharge);
    }
    let resulting_len = owner
        .frontiers
        .retained_records()
        .len()
        .checked_add(rows.len());
    if resulting_len
        .and_then(|len| u64::try_from(len).ok())
        .is_none_or(|len| len > owner.retained_record_limit)
    {
        return failure(owner, operation, LiveFrontierError::RetainedRecordLimit);
    }
    let Some(accounting) = accounting_after_rows(owner.closure_accounting, &charges) else {
        return failure(owner, operation, LiveFrontierError::ClosureAccounting);
    };
    owner.frontiers = match owner
        .frontiers
        .apply_live_transition(active, rows, sequence, order)
    {
        Ok(frontiers) => frontiers,
        Err(frontier_failure) => {
            let (frontiers, error) = *frontier_failure;
            owner.frontiers = frontiers;
            return failure(owner, operation, map_frontier_error(error));
        }
    };
    owner.retained_charges.extend(charges);
    owner.closure_accounting = accounting;
    Ok(LiveFrontierCommit { operation, owner })
}

const fn map_frontier_error(error: LiveFrontierTransitionError) -> LiveFrontierError {
    match error {
        LiveFrontierTransitionError::Authority => LiveFrontierError::Authority,
        LiveFrontierTransitionError::Precedence(condition) => {
            LiveFrontierError::Precedence(condition)
        }
        LiveFrontierTransitionError::RecordPosition
        | LiveFrontierTransitionError::Exhausted
        | LiveFrontierTransitionError::ResultingFrontier => LiveFrontierError::Frontier,
    }
}

fn failure<T, U>(
    owner: LiveFrontierOwner,
    operation: T,
    error: LiveFrontierError,
) -> Result<U, Box<LiveFrontierFailure<T>>> {
    Err(Box::new(LiveFrontierFailure {
        error,
        operation,
        owner,
    }))
}