liminal-server 0.8.0

//! Credential-attach arm of the production handler.
//!
//! Classification flows through the shared credential-attach lookup (token
//! phase, tombstone precedence, verifier order, live-authority checks),
//! commits through the crate's verified attach transitions — ordinary
//! detached attach or the R-C1.3 superseding handoff — mints the shell event
//! through the A3 aggregate commit, and answers through the request-bound
//! response authority. No lifecycle outcome is constructed here.
//!
//! Error contract: any [`StateError`] leaves durable state untouched (nothing
//! is published before the append succeeds) but may have consumed in-memory
//! authority. The handler therefore discards the whole in-memory conversation
//! owner on error and cold-replays durable reality on the next touch — the
//! same crash-consistency model the aggregate barrier is built for.

use std::sync::Arc;

use liminal::durability::DurableStore;
use liminal_protocol::lifecycle::{
    AggregateOperationDecision, AttachCommit, AttachFrontierCharges, AttachTransition,
    BindingSlotDecision, BindingState, CredentialAttachLiveReceipt, CredentialAttachLookupResult,
    LiveFrontierError, LiveFrontierOwner, PrecedenceCondition, PresentedIdentity, ReceiptDeadlines,
    RetainedRecordCharge, SemanticConnectionCapacityDecision, apply_attach_frontier, commit_attach,
    decide_attached_operation, lookup_credential_attach, select_credential_attach_binding_slot,
};
use liminal_protocol::wire::{
    AttachBound, AttachEnvelope, AttachSecret, BindingEpoch, CredentialAttachRequest,
    CredentialAttachResponse, Generation, ObserverBackpressureState, ReceiptExpiryReason,
};

use crate::server::participant::dispatch_impact::DispatchImpactAccumulator;

use super::barrier::{ArmOutcome, CommitMode, OperationFacts, commit_through_barrier};
use super::capacity::ServerCapacity;
use super::dispatch_impact::AttachedSourceRecord;
use super::facts::{self, Digest};
use super::frontier;
use super::log::{
    StoredAttachAllocation, StoredAttachModeV3, StoredAttachRequest, StoredOperation,
};
use super::non_presenting_finalizer::NonPresentingFinalizerCommit;
use super::observer_progress::ObserverProgressSourceMetadata;
use super::occupancy::ProvenanceMember;
use super::ops_attach_capacity::AttachStage8;
use super::ops_attach_finalizer::SelectedFencedFinalizer;
use super::ops_attach_lookup::{credential_attach_refusal, marker_bearing_attach_refusal};
use super::ops_attach_verify::{AttachVerification, stored_attach_parameters, verify_attach_mode};
use super::outbox_projection::capture_projection_prestate;
use super::presented_refusal::PresentedRefusal;
use super::state::{
    AttachProvenanceRecord, AttachReceiptState, ConversationAuthority, DurableAppend,
    PendingBindingFate, Slot, StateError,
};

/// Whether this commit path may PRESENT a §0.16 settlement refusal.
///
/// Live requests may; a cold replay may not — see `transition_attach_frontier`.
#[derive(Clone, Copy)]
enum AttachPresentation {
    /// A live request, carrying the observer progress condition 1's row needs.
    Live {
        /// Durable hard-observer progress, the initial refusal epoch.
        observer_progress: u64,
    },
    /// A durable row being replayed. Never presented.
    Replay,
}

impl ConversationAuthority {
    /// Applies one credential-attach request end to end.
    ///
    /// Attach never creates a conversation: a fresh conversation id has no
    /// slots and classifies as `ParticipantUnknown` without any durable
    /// append.
    pub(super) fn apply_credential_attach_with_impact(
        &mut self,
        request: &CredentialAttachRequest,
        operation_facts: &OperationFacts,
        server_capacity: &ServerCapacity,
        store: Arc<dyn DurableStore>,
        appender: &dyn DurableAppend,
        impact: &mut DispatchImpactAccumulator,
    ) -> Result<ArmOutcome, StateError> {
        let envelope = attach_envelope(request);
        let now = u128::from(operation_facts.now_ms);
        // Request-time expiry of retained provenance fingerprints (contract
        // R-C0: retained only through their provenance deadlines). Safe
        // before lookup: an expired record and a pruned record classify
        // identically through the generation-window witness.
        self.prune_expired_provenance(now);
        let Some(slot) = self.slots.get(&request.participant_id) else {
            return Ok(ArmOutcome::respond(
                CredentialAttachResponse::participant_unknown(envelope).into_server_value(),
            ));
        };
        let (token_phase, secret_proof) = slot.attach_token_phase(request, now);
        let lookup = lookup_credential_attach(
            token_phase,
            PresentedIdentity::Live(&slot.member),
            &slot.binding,
            request,
            secret_proof,
        );
        if !matches!(lookup, CredentialAttachLookupResult::AuthorizedFresh { .. }) {
            return credential_attach_refusal(&lookup, envelope, slot).map(ArmOutcome::respond);
        }
        // Stage 6, first half: connection-conversation capacity (register
        // row 5641) — after the lookup stages, before binding-slot occupancy,
        // the crate's frozen stage order.
        let capacity = match operation_facts.semantic_connection_capacity() {
            SemanticConnectionCapacityDecision::Commit(value) => value,
            SemanticConnectionCapacityDecision::Respond { limit } => {
                return Ok(ArmOutcome::respond(
                    CredentialAttachResponse::connection_conversation_capacity_exceeded(
                        envelope, limit,
                    )
                    .into_server_value(),
                ));
            }
        };
        if let BindingSlotDecision::Respond(response) = select_credential_attach_binding_slot(
            request,
            self.binding_slot_occupancy(operation_facts.receiving_incarnation),
        ) {
            return Ok(ArmOutcome::respond(response.into_server_value()));
        }
        // Decision D §5.3 permits the fenced mint only after an exact durable
        // source row and the owner-held `ValidatedMarkerRecord` have both been
        // validated; it expressly forbids a raw replacement from the request.
        // This live binding has no participant-record delivery pump yet, so its
        // factual delivery/source state is empty. Preserve that authority
        // boundary by returning the total selector's typed refusal until the
        // later delivery owner can supply those facts.
        if request.accept_marker_delivery_seq.is_some() {
            return marker_bearing_attach_refusal(self, request, slot, operation_facts)
                .map(ArmOutcome::respond);
        }
        // Stage 8 (R-D1): credential attach's exact five-scope
        // receipt/provenance order, decided through the crate's verified
        // selector against per-participant/per-conversation occupancies from
        // this authority and server occupancies from the shared ledger; the
        // reservation is atomic with the check.
        let deadlines = operation_facts.deadlines()?;
        let (reservation, retire, displace) = match self.attach_stage8(
            request,
            slot,
            operation_facts,
            server_capacity,
            &deadlines,
        )? {
            AttachStage8::Refused(response) => {
                return Ok(ArmOutcome::respond(response.into_server_value()));
            }
            AttachStage8::Reserved {
                reservation,
                retire,
                displace,
            } => (reservation, retire, displace),
        };
        // §0.16 obligation 1. Captured BEFORE the first allocator runs, so a
        // settlement refusal raised four calls deeper can put the allocators
        // back — the `Ok` at the funnel retains this owner instead of cold
        // replaying it, so nothing else ever would.
        let captured_positions = self.position_allocators();
        let (attached_order, attached_seq, attach_mode) =
            self.allocate_attach_mode(slot.binding, envelope)?;
        let allocation = stored_attach_allocation(
            request,
            operation_facts,
            &deadlines,
            attached_order,
            attached_seq,
        )?;
        let source_log_sequence = self.next_log_sequence;
        let source = attached_source(request, allocation, &attach_mode);
        let projection_facts = capture_projection_prestate(self, &source);
        let outcome = match self.attach_commit(
            request,
            &allocation,
            &attach_mode,
            store,
            CommitMode::Live(appender),
        ) {
            Ok(outcome) => outcome,
            Err(error) => {
                if matches!(error, StateError::PresentedRefusal(_)) {
                    self.restore_position_allocators(captured_positions);
                }
                return Err(error);
            }
        };
        self.record_attached_source(
            AttachedSourceRecord {
                source_log_sequence,
                source: &source,
                projection_facts,
                participant_id: request.participant_id,
            },
            appender,
            impact,
        )?;
        // The durable append succeeded: the stage-8 reservation becomes
        // permanent and the receipts this rotation retired early (the
        // superseded attach receipt and, on the first rotation, the ended
        // enrollment receipt) leave the server-scope ledger — together with
        // the provenance entries the participant's own window displaced.
        reservation.confirm(&retire, &displace);
        Ok(ArmOutcome::committed(
            CredentialAttachResponse::attach_bound(outcome).into_server_value(),
            capacity,
        ))
    }

    /// Selects the mandatory v3 Attached mode from exact binding authority and
    /// consumes only the matching checked order/sequence allocation.
    ///
    /// # The `PendingFinalization` arm is a REFUSAL, not an invariant
    ///
    /// Board #14. This arm used to be a bare `StateError::invariant` whose
    /// message — "pending finalization observed in a binding that commits
    /// detaches immediately" — asserted the state could not arise. It arises:
    /// a well-formed, correctly-authorized, current-generation attach reaches
    /// it (`tests_14_attach_presentation`), and board #23 reaches it a second
    /// way through a connection dropped under retention pressure. The verb is
    /// the contract and the comment was not.
    ///
    /// **The refusal itself does not change and must not.** A binding whose
    /// terminal has not yet been appended cannot admit an attach, and
    /// `docs/design/ATTACH-SILENCE-14.md`'s acceptance constraint spells out
    /// why merely admitting it would be worse than the silence: the attach
    /// would be COMMITTED AND STORED, and on the next cold replay it would
    /// arrive at `select_fenced_finalizer` through `replay_attached`, a path
    /// this gate does not cover, and meet three more bare-close sinks. So the
    /// state stays refused, nothing is committed, and only the DELIVERY of the
    /// refusal changes.
    ///
    /// # Why `ObserverBackpressure` is the row, and not a near neighbour
    ///
    /// `PendingFinalization` is minted in exactly one circumstance. The
    /// protocol's `BindingTerminalAdmission::Pending` arm is reached only when
    /// `hard_observer_progress < key.delivery_seq`, and its own type documents
    /// itself as the "observer-blocked pending terminal admission". The
    /// contract says the same from the other side: a durably
    /// `PendingFinalization` slot is settled when "progress wake appends
    /// exactly one correctly ordered record" (`PARTICIPANT-CONTRACT.md`), and
    /// the register pairs the detach that CREATES the state with
    /// `ObserverBackpressure` (the "first accepted while append is blocked"
    /// row). The blocked resource is hard-observer progress; the wake that
    /// clears it is `ObserverProgressed`; and `ObserverBackpressure` is an
    /// outcome the register already admits for credential attach, carrying
    /// exactly the retry discipline "retry once after matching
    /// `ObserverProgressed`". Nothing is invented — the row was always the
    /// right one and the refusal simply never reached it.
    ///
    /// The state is `initial(observer_progress)`, whose doc reads "an initial
    /// refusal epoch is exactly the progress value observed by the serialized
    /// operation". `replay(..)` is the exact-token detach-replay form and is
    /// not this shape.
    ///
    /// ⚠ One honest limit. R-D1's stage order puts `ObserverBackpressure`
    /// (stage 11) after `ConversationOrderExhausted` (9) and
    /// `ConversationSequenceExhausted` (10), and this arm refuses before
    /// `allocate_position` runs — so a request that is BOTH against a pending
    /// binding AND at an exhausted order/sequence would now hear the stage-11
    /// row where the register wants the stage-9 or stage-10 one. That
    /// inversion is inherited, not introduced: the pre-#14 code refused at the
    /// same point, one line earlier in the same arm. Recorded rather than
    /// silently absorbed.
    fn allocate_attach_mode(
        &mut self,
        binding: BindingState,
        envelope: AttachEnvelope,
    ) -> Result<(u64, u64, StoredAttachModeV3), StateError> {
        match binding {
            BindingState::Detached => {
                let (order, sequence) = self.allocate_position()?;
                Ok((order, sequence, StoredAttachModeV3::Ordinary))
            }
            BindingState::Bound(active) => {
                let (order, terminal_sequence, attached_sequence) =
                    self.allocate_supersession_position()?;
                Ok((
                    order,
                    attached_sequence,
                    StoredAttachModeV3::Superseding {
                        prior_binding_epoch: active.binding_epoch.into(),
                        terminal_transaction_order: order,
                        terminal_delivery_seq: terminal_sequence,
                    },
                ))
            }
            BindingState::PendingFinalization(_) => Err(StateError::PresentedRefusal(
                PresentedRefusal::credential_attach(
                    CredentialAttachResponse::observer_backpressure(
                        envelope,
                        ObserverBackpressureState::initial(self.observer_progress),
                    ),
                ),
            )),
        }
    }

    /// Whether this commit path may PRESENT a §0.16 settlement refusal.
    const fn attach_presentation(&self, mode: &CommitMode<'_>) -> AttachPresentation {
        match mode {
            CommitMode::Live(_) => AttachPresentation::Live {
                observer_progress: self.observer_progress,
            },
            CommitMode::Replay { .. } => AttachPresentation::Replay,
        }
    }

    /// Restores every authority `attach_commit` consumed before the refusal
    /// existed, then presents it (participant contract §0.16, build
    /// obligation 1).
    ///
    /// # What must be restored, and why `into_parts` alone is not enough
    ///
    /// The refusal is raised inside `attach_commit`, AFTER `slots.remove_entry`,
    /// `take_frontier`, and `prepare_selected_fenced_finalizer` have run: the
    /// carrier consumes authority before the refusal exists, and
    /// `LiveFrontierFailure::into_parts` gives back only the frontier owner.
    /// So this restores the slot entry and the frontier here, the arm restores
    /// the position allocators (see [`PositionAllocators`]), and the pin
    /// measures all three plus the finalizer state rather than asserting them.
    ///
    /// # ⛔ The finalizer tripwire
    ///
    /// A prepared fenced finalizer has NO inverse: `select_fenced_finalizer`
    /// has already consumed a fate-occurrence presentation owner and
    /// `prepare_selected_fenced_finalizer` has moved a pending specific fate
    /// into a prepared ordinary one. Rather than half-restore, this path
    /// REFUSES TO PRESENT when a finalizer was selected and falls back to the
    /// pre-amendment bare close, which discards the part-consumed owner and
    /// cold-replays durable truth. That arm is unreachable in production today
    /// — the live path refuses `PendingFinalization` one stage earlier at
    /// `allocate_attach_mode`, so `select_fenced_finalizer` returns `None` —
    /// and it is exactly §0.16 condition 3's census-excluded fenced territory.
    /// The first production `StoredAttachModeV3::Fenced` constructor makes this
    /// arm live and owes the restore its own lane.
    fn restore_and_present_attach(
        &mut self,
        participant_id: liminal_protocol::wire::ParticipantId,
        slot: Slot,
        owner: LiveFrontierOwner,
        finalizer_selected: bool,
        response: CredentialAttachResponse,
    ) -> Result<AttachBound, StateError> {
        if finalizer_selected {
            return Err(StateError::invariant(
                "attach frontier transition failed: a settlement refusal cannot be presented \
                 after a fenced finalizer was selected -- see restore_and_present_attach",
            ));
        }
        self.install_frontier(owner)?;
        self.slots.insert(participant_id, slot);
        Err(StateError::PresentedRefusal(
            PresentedRefusal::credential_attach(response),
        ))
    }

    /// Replays one committed attach entry from its stored inputs.
    pub(super) fn replay_attached(
        &mut self,
        request: StoredAttachRequest,
        allocation: &StoredAttachAllocation,
        attach_mode: &StoredAttachModeV3,
        stored_event: &[u8],
        sequence: u64,
        store: Arc<dyn DurableStore>,
    ) -> Result<(), StateError> {
        let request = request.to_request()?;
        self.attach_commit(
            &request,
            allocation,
            attach_mode,
            store,
            CommitMode::Replay {
                stored_event,
                sequence,
            },
        )?;
        Ok(())
    }

    /// Shared credential-attach commit core (live and replay paths).
    ///
    /// The mode is derived from the slot's binding authority paired with the
    /// stored allocation: a detached slot with no terminal allocation binds
    /// ordinarily; a bound slot with a terminal allocation supersedes its
    /// active epoch atomically (one ordered `Detached(Superseded)`/`Attached`
    /// handoff through the crate's verified transition). Any other pairing is
    /// a drifted log and fails loudly.
    #[expect(
        clippy::too_many_lines,
        reason = "the amendment adds ONE conditional arm (the settlement refusal and its \
                  restoration) to an already-long commit core; splitting the core to fit the \
                  lint would move the take/install pairing of the frontier, the slot, and the \
                  shell across a function boundary, which is exactly the coupling the \
                  restoration obligation depends on being visible in one place"
    )]
    pub(super) fn attach_commit(
        &mut self,
        request: &CredentialAttachRequest,
        allocation: &StoredAttachAllocation,
        attach_mode: &StoredAttachModeV3,
        store: Arc<dyn DurableStore>,
        mode: CommitMode<'_>,
    ) -> Result<AttachBound, StateError> {
        let source_sequence = self.next_log_sequence;
        let presentation = self.attach_presentation(&mode);
        let (participant_id, mut slot) = self
            .slots
            .remove_entry(&request.participant_id)
            .ok_or_else(|| {
                StateError::invariant("attach commit requires an enrolled participant slot")
            })?;
        let finalizer = self.select_fenced_finalizer(slot.binding, attach_mode, request)?;
        let non_presenting = finalizer.is_some_and(SelectedFencedFinalizer::is_non_presenting);
        let live_appender = match mode {
            CommitMode::Live(appender) => Some(appender),
            CommitMode::Replay { .. } => None,
        };
        let (result_generation, parameters) = stored_attach_parameters(request, allocation)?;
        let frontier_owner = self.take_frontier()?;
        let (verified, frontier_owner) = verify_attach_mode(
            // Cloned rather than moved: §0.16 obligation 1 needs the WHOLE slot
            // entry intact if the frontier transition below presents a
            // settlement refusal, and a partial move leaves nothing to restore.
            slot.member.clone(),
            slot.binding,
            frontier_owner,
            AttachVerification {
                request,
                mode: attach_mode,
                parameters,
                store,
                source_sequence,
            },
        )?;
        let (frontier_owner, completes_ordinary) = self.prepare_selected_fenced_finalizer(
            request.participant_id,
            source_sequence,
            finalizer,
            frontier_owner,
        )?;
        let committed = commit_attach(verified, slot.cell).map_err(|error| {
            StateError::invariant(format!("protocol attach transition failed: {error:?}"))
        })?;
        let observer_projection = if non_presenting {
            None
        } else {
            committed.observer_progress_projection()
        };
        let transitioned = transition_attach_frontier(
            frontier_owner,
            committed,
            request,
            allocation,
            presentation,
        )?;
        let (committed, frontier_owner) = match transitioned {
            AttachFrontierOutcome::Committed(committed, owner) => (committed, owner),
            AttachFrontierOutcome::Presented(response, owner) => {
                return self.restore_and_present_attach(
                    participant_id,
                    slot,
                    owner,
                    finalizer.is_some(),
                    *response,
                );
            }
        };
        let (committed, frontier_owner) = if non_presenting {
            NonPresentingFinalizerCommit::new(committed, frontier_owner).into_parts()
        } else {
            (committed, frontier_owner)
        };
        let shell = self.take_shell()?;
        let barrier = match decide_attached_operation(shell, committed) {
            AggregateOperationDecision::Commit(barrier) => barrier,
            AggregateOperationDecision::Refused(refusal) => {
                return Err(StateError::ShellRefused {
                    reason: refusal.reason(),
                });
            }
        };
        let make_operation = |event: Vec<u8>| StoredOperation::Attached {
            request: request.into(),
            secret_verified: true,
            allocation: *allocation,
            mode: Box::new(attach_mode.clone()),
            event,
        };
        let (shell, committed) =
            commit_through_barrier(barrier, mode, self.next_log_sequence, &make_operation)?;
        self.shell = Some(shell);
        self.install_frontier(frontier_owner)?;
        self.advance_log_head()?;
        let outcome = committed.outcome.clone();
        let (installed, fate_token) = committed.into_slot_and_fate();
        slot.member = installed.member;
        slot.binding = installed.binding_state;
        slot.binding_fate = Some(PendingBindingFate {
            attached_source_sequence: source_sequence,
            token: fate_token,
        });
        slot.cell = installed.detach_cell;
        slot.attach_secret = AttachSecret::new(allocation.attach_secret);
        install_attach_receipt(
            &mut slot,
            request,
            allocation,
            &outcome,
            installed.outcome,
            result_generation,
            self.receipt_limits.provenance_participant_window,
        );
        self.slots.insert(participant_id, slot);
        if let Some(projection) = observer_projection {
            let terminal_delivery_seq = projection.new_observer_progress();
            let metadata = attach_metadata(source_sequence, request, terminal_delivery_seq);
            self.record_observer_progress_projection(projection, metadata)?;
        }
        self.observe_replayed_position(allocation.attached_order, allocation.attached_seq)?;
        if completes_ordinary && let Some(appender) = live_appender {
            self.complete_prepared_ordinary_finalizer(request.participant_id, appender)?;
        }
        Ok(outcome)
    }
}

fn install_attach_receipt(
    slot: &mut Slot,
    request: &CredentialAttachRequest,
    allocation: &StoredAttachAllocation,
    outcome: &AttachBound,
    installed_outcome: AttachBound,
    result_generation: Generation,
    provenance_window: u64,
) {
    // Lane p0-39: plan the participant's window displacement against the
    // PRE-commit slot, before any of the mutations below move it. This is the
    // same call `attach_stage8` made to build the ledger's removal list, over
    // the same state, so the ledger and the slot cannot retain different sets —
    // and because the plan reads only durable structure and the configured
    // window (never the clock), a cold replay of these same commits in the same
    // order re-derives the identical retained set.
    let displaced = slot
        .incoming_provenance_member()
        .map_or_else(Vec::new, |incoming| {
            slot.plan_provenance_displacement(incoming, provenance_window)
        });
    // Retire the previous receipt into its bounded provenance record with the
    // exact terminal reason: `Superseded` when the newer generation ended a
    // still-live receipt, `Deadline` when its own deadline had already ended
    // it. The admitted clock makes replay reproduce the identical record.
    if let Some(previous) = slot.attach.take() {
        let reason = if u128::from(allocation.admitted_now_ms) < previous.receipt_expires_at {
            ReceiptExpiryReason::Superseded
        } else {
            ReceiptExpiryReason::Deadline
        };
        // The retired predecessor IS this commit's incoming window member; a
        // window too small to hold even that displaces it immediately, so it
        // is never retained in the first place.
        let member = ProvenanceMember::Attach {
            expires_at: previous.provenance_expires_at,
            token: previous.token.into_bytes(),
        };
        if !displaced.contains(&member) {
            slot.attach_provenance.insert(
                previous.token.into_bytes(),
                AttachProvenanceRecord {
                    result_generation: previous.result_generation,
                    reason,
                    provenance_expires_at: previous.provenance_expires_at,
                },
            );
        }
    }
    // The first rotation also ends the enrollment receipt's secret body. Set
    // once and never rewrite it, preserving the exact end-of-body fact.
    if slot.enrollment_receipt_ended.is_none() {
        slot.enrollment_receipt_ended = Some(
            if u128::from(allocation.admitted_now_ms) < slot.enrollment_receipt_expires_at {
                ReceiptExpiryReason::Superseded
            } else {
                ReceiptExpiryReason::Deadline
            },
        );
    }
    // Apply the plan: the window drops its oldest members so the newest one of
    // the same participant can land.
    for member in displaced {
        match member {
            ProvenanceMember::Enrollment { .. } => {
                slot.enrollment_provenance_displaced = true;
            }
            ProvenanceMember::Attach { token, .. } => {
                slot.attach_provenance.remove(&token);
            }
        }
    }
    slot.attach = Some(AttachReceiptState {
        token: request.attach_attempt_token,
        receipt: CredentialAttachLiveReceipt::from_commit(outcome.clone()),
        outcome: installed_outcome,
        verifier: request.attach_secret.into_bytes(),
        result_generation,
        receipt_expires_at: allocation.receipt_expires_at.get(),
        provenance_expires_at: allocation.provenance_expires_at.get(),
    });
}

const fn attach_metadata(
    source_sequence: u64,
    request: &CredentialAttachRequest,
    terminal_delivery_seq: u64,
) -> ObserverProgressSourceMetadata {
    ObserverProgressSourceMetadata::attached(
        source_sequence,
        request.conversation_id,
        request.participant_id,
        terminal_delivery_seq,
    )
}

/// Outcome of coupling one sealed attach commit to the live frontier.
///
/// The refusing arm exists because §0.16's law attaches to the SEAM: the
/// wrapper now reports WHICH clearing condition blocked, and two of the three
/// have a lawful wire answer. It hands the UNCHANGED owner straight back so the
/// caller can restore every consumed authority before the refusal exists.
#[expect(
    clippy::large_enum_variant,
    reason = "the committed arm carries the whole attach commit and its owner by design -- the \
              refusing arm is boxed already, and boxing the hot committed arm would pay an \
              allocation on every successful attach to shrink a value that is moved once"
)]
enum AttachFrontierOutcome {
    /// The transition entered the frontier.
    Committed(AttachCommit<Digest, Digest>, LiveFrontierOwner),
    /// Participant contract §0.16: a lawful refusal the register admits for
    /// credential attach, paired with the intact owner it must be restored
    /// alongside.
    Presented(Box<CredentialAttachResponse>, LiveFrontierOwner),
}

/// Selects the lawful §0.16 answer for a clearing condition, or `None` where the
/// amendment rules no row.
///
/// Condition 1 reuses the register's existing `ObserverBackpressure` row (the
/// blocked resource IS hard-observer progress, and `ObserverProgressed` is a
/// wake something actually sends). Condition 2 mints the settlement row with
/// the epoch the frontier itself named. Condition 3 — armed fenced recovery —
/// is EXCLUDED BY CENSUS (board #13: zero production constructors of
/// `StoredAttachModeV3::Fenced`), so it keeps the pre-amendment bare close and
/// its tripwire; `Unclassified` keeps it for the same reason from the other
/// side, having never been one of the amendment's conditions at all.
const fn attach_settlement_answer(
    condition: PrecedenceCondition,
    envelope: AttachEnvelope,
    observer_progress: u64,
) -> Option<CredentialAttachResponse> {
    match condition {
        PrecedenceCondition::BindingTerminal => {
            Some(CredentialAttachResponse::observer_backpressure(
                envelope,
                ObserverBackpressureState::initial(observer_progress),
            ))
        }
        PrecedenceCondition::MarkerDrain { settlement_epoch } => Some(
            CredentialAttachResponse::marker_settlement_backpressure(&envelope, settlement_epoch),
        ),
        // ⛔ TRIPWIRE (§0.16 condition 3). The first production constructor of
        // `StoredAttachModeV3::Fenced` VOIDS the census exclusion, and the row
        // question reopens as a blocking prerequisite of that constructor's own
        // lane — this exclusion may never be cited to land the constructor
        // without the row. Until then the arm is unreachable in production and
        // a refusal row for it would be wire surface for nothing.
        PrecedenceCondition::FencedRecovery | PrecedenceCondition::Unclassified => None,
    }
}

fn transition_attach_frontier(
    owner: LiveFrontierOwner,
    committed: AttachCommit<Digest, Digest>,
    request: &CredentialAttachRequest,
    allocation: &StoredAttachAllocation,
    presentation: AttachPresentation,
) -> Result<AttachFrontierOutcome, StateError> {
    let attached_encoded = frontier::credential_attached_charge(
        request.conversation_id,
        request.participant_id,
        allocation,
    )?;
    let attached_charge = RetainedRecordCharge::new(
        committed.attached.delivery_seq(),
        committed.attached.admission_order(),
        attached_encoded,
    );
    let terminal = match committed.transition {
        AttachTransition::Detached => None,
        AttachTransition::Superseded { terminal } => Some(terminal.into()),
        AttachTransition::FencedRecovery {
            composed_terminal, ..
        } => composed_terminal,
    };
    let terminal_charge = terminal
        .map(|terminal| {
            frontier::terminal_charge(
                terminal.conversation_id(),
                terminal.participant_id(),
                terminal.binding_epoch(),
                terminal.admission_order().transaction_order(),
                terminal.delivery_seq(),
            )
            .map(|encoded| {
                RetainedRecordCharge::new(
                    terminal.delivery_seq(),
                    terminal.admission_order(),
                    encoded,
                )
            })
        })
        .transpose()?;
    match apply_attach_frontier(
        owner,
        committed,
        AttachFrontierCharges::new(terminal_charge, attached_charge),
    ) {
        Ok(commit) => {
            let (committed, owner) = commit.into_parts();
            Ok(AttachFrontierOutcome::Committed(committed, owner))
        }
        Err(failure) => {
            let error = failure.error();
            // ⛔ REPLAY IS NEVER PRESENTED. A durable row that replays into a
            // `Precedence` refusal is a drifted log, not backpressure — there is
            // no client waiting on it and answering "retry later" to a cold
            // replay would convert a state defect into a silent success.
            let answer = match (presentation, error) {
                (
                    AttachPresentation::Live { observer_progress },
                    LiveFrontierError::Precedence(condition),
                ) => {
                    attach_settlement_answer(condition, attach_envelope(request), observer_progress)
                }
                _ => None,
            };
            answer.map_or_else(
                || {
                    Err(StateError::invariant(format!(
                        "attach frontier transition failed: {error:?}"
                    )))
                },
                |response| {
                    let (_, owner) = failure.into_parts();
                    Ok(AttachFrontierOutcome::Presented(Box::new(response), owner))
                },
            )
        }
    }
}

/// Builds the echo envelope of one credential-attach request.
pub(super) const fn attach_envelope(request: &CredentialAttachRequest) -> AttachEnvelope {
    AttachEnvelope {
        conversation_id: request.conversation_id,
        participant_id: request.participant_id,
        capability_generation: request.capability_generation,
        attach_attempt_token: request.attach_attempt_token,
        accept_marker_delivery_seq: request.accept_marker_delivery_seq,
    }
}

fn attached_source(
    request: &CredentialAttachRequest,
    allocation: StoredAttachAllocation,
    mode: &StoredAttachModeV3,
) -> StoredOperation {
    StoredOperation::Attached {
        request: request.into(),
        secret_verified: true,
        allocation,
        mode: Box::new(mode.clone()),
        event: Vec::new(),
    }
}

/// Mints one committed attach allocation from the rotation's exact inputs.
///
/// The rotation result carries the successor of the verified current
/// generation (the crate's `ResultGeneration` law), and both deadlines come from
/// the receipt window this operation already decided — never from the request.
fn stored_attach_allocation(
    request: &CredentialAttachRequest,
    operation_facts: &OperationFacts,
    deadlines: &ReceiptDeadlines,
    attached_order: u64,
    attached_seq: u64,
) -> Result<StoredAttachAllocation, StateError> {
    let next_generation = request
        .capability_generation
        .get()
        .checked_add(1)
        .and_then(Generation::new)
        .ok_or(StateError::AllocationExhausted {
            domain: "capability generation",
        })?;
    Ok(StoredAttachAllocation {
        binding_epoch: BindingEpoch::new(operation_facts.receiving_incarnation, next_generation)
            .into(),
        attach_secret: facts::mint_secret_bytes()?,
        attached_order,
        attached_seq,
        receipt_expires_at: deadlines.receipt_expires_at().into(),
        provenance_expires_at: deadlines.provenance_expires_at().into(),
        admitted_now_ms: operation_facts.now_ms,
    })
}