liminal_protocol/client/

inbound.rs

use super::{
    ClientBindingState, ClientParticipantAggregate, ClientResponseCorrelation, correlation,
};
use crate::wire::{AttachBound, ReceiptReplay, ServerValue};

/// Closed refusal classes for inbound semantic values.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ClientInboundRefusalReason {
    /// A durable Leave already terminalized the local participant.
    AlreadyDead,
    /// The value names another operation or participant identity.
    ForeignResponse,
    /// The value is absent an expectation or belongs to an older request.
    DelayedResponse,
    /// Wire identity is insufficient to assign this value to one expected operation.
    AmbiguousResponse,
    /// An expected-operation response was presented without its one-use send correlation.
    MissingResponseAuthority,
    /// A restore already testified the issued send authority destroyed; only
    /// the pending testimony can resolve the operation (r2, 2026-07-18).
    LostAuthorityPending,
}

/// Applied inbound value and resulting aggregate.
#[derive(Debug, PartialEq, Eq)]
pub struct ClientInboundApplied {
    aggregate: ClientParticipantAggregate,
    value: ServerValue,
}

impl ClientInboundApplied {
    /// Releases the resulting aggregate and exact applied value.
    #[must_use]
    pub fn into_parts(self) -> (ClientParticipantAggregate, ServerValue) {
        (self.aggregate, self.value)
    }
}

/// Refused inbound value paired with the unchanged aggregate.
#[derive(Debug, PartialEq, Eq)]
pub struct ClientInboundRefusal {
    aggregate: ClientParticipantAggregate,
    value: ServerValue,
    reason: ClientInboundRefusalReason,
}

impl ClientInboundRefusal {
    /// Returns the closed refusal reason.
    #[must_use]
    pub const fn reason(&self) -> ClientInboundRefusalReason {
        self.reason
    }

    /// Releases the unchanged aggregate and exact refused value.
    #[must_use]
    pub fn into_parts(self) -> (ClientParticipantAggregate, ServerValue) {
        (self.aggregate, self.value)
    }
}

/// Exhaustive inbound correlation decision.
#[derive(Debug, PartialEq, Eq)]
pub enum ClientInboundDecision {
    /// The crate correlated and applied the typed value.
    Applied(ClientInboundApplied),
    /// The crate retained both authority and value unchanged.
    Refused(ClientInboundRefusal),
}

/// Refused body-omitting response with the exact local correlation retained.
#[derive(Debug, PartialEq, Eq)]
pub struct ClientCorrelatedInboundRefusal {
    aggregate: ClientParticipantAggregate,
    value: ServerValue,
    correlation: ClientResponseCorrelation,
    reason: ClientInboundRefusalReason,
}

impl ClientCorrelatedInboundRefusal {
    /// Returns the closed refusal reason.
    #[must_use]
    pub const fn reason(&self) -> ClientInboundRefusalReason {
        self.reason
    }

    /// Releases every unchanged input, including the non-cloneable correlation.
    #[must_use]
    pub fn into_parts(
        self,
    ) -> (
        ClientParticipantAggregate,
        ServerValue,
        ClientResponseCorrelation,
    ) {
        (self.aggregate, self.value, self.correlation)
    }
}

/// Applied inbound value that did not answer the outstanding operation.
///
/// The value applied on its own authority -- a continuous acknowledgement
/// answer or a pre-semantic transport rejection -- and left the expected slot
/// standing under the same issued authorization. The actual answer is still
/// owed, so the process-local correlation comes back with the aggregate
/// instead of being spent.
#[derive(Debug, PartialEq, Eq)]
pub struct ClientCorrelatedInboundRetained {
    aggregate: ClientParticipantAggregate,
    value: ServerValue,
    correlation: ClientResponseCorrelation,
}

impl ClientCorrelatedInboundRetained {
    /// Releases the resulting aggregate, the applied value, and the still-current correlation.
    #[must_use]
    pub fn into_parts(
        self,
    ) -> (
        ClientParticipantAggregate,
        ServerValue,
        ClientResponseCorrelation,
    ) {
        (self.aggregate, self.value, self.correlation)
    }
}

/// Inbound decision for response classes whose wire envelopes omit request identity.
#[derive(Debug, PartialEq, Eq)]
pub enum ClientCorrelatedInboundDecision {
    /// The exact local operation authorization and wire envelope both matched;
    /// the correlation is spent.
    Applied(ClientInboundApplied),
    /// The value applied without answering the outstanding operation; the
    /// correlation is handed back because the answer is still owed.
    AppliedRetaining(ClientCorrelatedInboundRetained),
    /// Aggregate, value, and correlation were retained unchanged.
    Refused(ClientCorrelatedInboundRefusal),
}

/// Correlates and applies one server value inside the client aggregate.
#[must_use]
pub fn decide_inbound(
    aggregate: ClientParticipantAggregate,
    value: ServerValue,
) -> ClientInboundDecision {
    decide_inbound_inner(aggregate, value, false)
}

/// Attempts correlation while retaining the process-local handle on refusal.
#[must_use]
pub fn decide_correlated_inbound(
    aggregate: ClientParticipantAggregate,
    value: ServerValue,
    correlation: ClientResponseCorrelation,
) -> ClientCorrelatedInboundDecision {
    let current_authority = aggregate.expected.as_ref().is_some_and(|expected| {
        expected.issued && expected.authorization == correlation.authorization
    });
    if !current_authority {
        return ClientCorrelatedInboundDecision::Refused(ClientCorrelatedInboundRefusal {
            aggregate,
            value,
            correlation,
            reason: ClientInboundRefusalReason::DelayedResponse,
        });
    }
    match decide_inbound_inner(aggregate, value, true) {
        ClientInboundDecision::Applied(applied) => {
            // An OBSERVATION of the applied aggregate, not a list of value
            // classes: whatever the value was, if the expected slot still
            // stands under this correlation's issued authorization the answer
            // is still owed and the correlation goes back with the aggregate.
            let still_owed = applied.aggregate.expected.as_ref().is_some_and(|expected| {
                expected.issued && expected.authorization == correlation.authorization
            });
            if still_owed {
                ClientCorrelatedInboundDecision::AppliedRetaining(ClientCorrelatedInboundRetained {
                    aggregate: applied.aggregate,
                    value: applied.value,
                    correlation,
                })
            } else {
                ClientCorrelatedInboundDecision::Applied(applied)
            }
        }
        ClientInboundDecision::Refused(refusal) => {
            let reason = refusal.reason();
            let (aggregate, value) = refusal.into_parts();
            ClientCorrelatedInboundDecision::Refused(ClientCorrelatedInboundRefusal {
                aggregate,
                value,
                correlation,
                reason,
            })
        }
    }
}

fn decide_inbound_inner(
    mut aggregate: ClientParticipantAggregate,
    value: ServerValue,
    has_response_authority: bool,
) -> ClientInboundDecision {
    if aggregate.binding.is_left() {
        return inbound_refusal(aggregate, value, ClientInboundRefusalReason::AlreadyDead);
    }

    if let Some(request) = correlation::participant_ack_request(&value) {
        if aggregate.binding.matches_ack(request) {
            return ClientInboundDecision::Applied(ClientInboundApplied { aggregate, value });
        }
        return inbound_refusal(
            aggregate,
            value,
            ClientInboundRefusalReason::ForeignResponse,
        );
    }

    if matches!(value, ServerValue::ParticipantTransportRejected(_)) {
        return ClientInboundDecision::Applied(ClientInboundApplied { aggregate, value });
    }

    let Some(expected) = aggregate.expected.as_ref() else {
        return inbound_refusal(
            aggregate,
            value,
            ClientInboundRefusalReason::DelayedResponse,
        );
    };

    if expected.lost.is_some() {
        return inbound_refusal(
            aggregate,
            value,
            ClientInboundRefusalReason::LostAuthorityPending,
        );
    }

    if !has_response_authority {
        return inbound_refusal(
            aggregate,
            value,
            ClientInboundRefusalReason::MissingResponseAuthority,
        );
    }

    if !aggregate.binding.accepts_request(&expected.request) {
        return inbound_refusal(
            aggregate,
            value,
            ClientInboundRefusalReason::ForeignResponse,
        );
    }

    if !correlation::matches_request(&value, &expected.request) {
        let same_request_class = value.originating_request()
            == Some(expected.request.discriminant())
            || matches!(
                (&value, &expected.request),
                (
                    ServerValue::ObserverRecoveryAccepted(_)
                        | ServerValue::InvalidObserverEpoch(_)
                        | ServerValue::InvalidObserverEpochList(_),
                    crate::wire::ClientRequest::ObserverRecovery(_)
                )
            );
        let same_identity = correlation::same_identity(&value, &expected.request);
        let reason = if same_request_class && same_identity {
            if matches!(
                expected.request,
                crate::wire::ClientRequest::RecordAdmission(_)
            ) {
                ClientInboundRefusalReason::AmbiguousResponse
            } else {
                ClientInboundRefusalReason::DelayedResponse
            }
        } else {
            ClientInboundRefusalReason::ForeignResponse
        };
        return inbound_refusal(aggregate, value, reason);
    }

    // A write the server could not settle answers nothing: the request may or
    // may not have committed. The expected slot stands under its authority,
    // so a correlated read hands the correlation back, and its holder ends the
    // expectation as a lost response and re-presents the same request with the
    // same attempt token.
    if matches!(value, ServerValue::WriteOutcomeUnknown(_)) {
        return ClientInboundDecision::Applied(ClientInboundApplied { aggregate, value });
    }

    retire_expected_operation(&mut aggregate, &value);
    ClientInboundDecision::Applied(ClientInboundApplied { aggregate, value })
}

/// Retires the expected slot and settles the detach replay in one statement.
///
/// The two are a coupled pair: an active replay without its exact expected
/// detach is a state `resume_record` refuses to encode
/// (`ClientResumeRecordEncodeError::DecoupledDetachReplay`) and no restore
/// accepts. So the clear may not happen on its own.
///
/// This function used to be that bare clear, under a comment asserting every
/// matching arm of `apply_correlated_value` supersedes or terminalizes the
/// replay. The assertion was false. A correlated value reaches here only when
/// its wire identity resolves to the expected request's key, and six values
/// whose key is a detach refuse that detach's AUTHORITY without answering it --
/// a rotated generation, a dropped binding, an unknown participant, a
/// connection-capacity or observer-backpressure refusal, a retirement older
/// than the replay. All six fell through `apply_correlated_value` untouched and
/// left the replay active behind a cleared slot.
///
/// The mechanism is deliberately an OBSERVATION rather than a longer list of
/// arms: whatever `apply_correlated_value` did or did not do, a replay left
/// active is settled here into a terminal retaining the exact refusing value.
/// A list is what was wrong before, and the list assembled while fixing it was
/// itself wrong in both directions -- it named a value that cannot correlate to
/// a detach (`ConversationOrderExhausted`, whose envelope has no detach
/// variant) and missed one that can (`ObserverBackpressure::Detach`). A new
/// correlatable refusal class therefore cannot reintroduce the decoupling,
/// because nothing here enumerates the classes.
fn retire_expected_operation(aggregate: &mut ClientParticipantAggregate, value: &ServerValue) {
    let expected_detach = match aggregate
        .expected
        .as_ref()
        .map(|expected| &expected.request)
    {
        Some(crate::wire::ClientRequest::Detach(request)) => Some(request.clone()),
        _ => None,
    };
    aggregate.expected = None;
    apply_correlated_value(aggregate, value);
    if let Some(request) = expected_detach {
        let envelope = crate::wire::DetachEnvelope {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation,
            detach_attempt_token: request.detach_attempt_token,
        };
        aggregate
            .detach_replay
            .settle_refused_authority(&envelope, value);
        debug_assert!(
            !aggregate.detach_replay.is_active()
                || aggregate.detach_replay.request() != Some(&envelope),
            "retiring an expected detach must leave its own replay settled"
        );
    }
    debug_assert!(
        aggregate.expected.is_none(),
        "the expected slot must be retired by this statement"
    );
}

const fn inbound_refusal(
    aggregate: ClientParticipantAggregate,
    value: ServerValue,
    reason: ClientInboundRefusalReason,
) -> ClientInboundDecision {
    ClientInboundDecision::Refused(ClientInboundRefusal {
        aggregate,
        value,
        reason,
    })
}

fn apply_correlated_value(aggregate: &mut ClientParticipantAggregate, value: &ServerValue) {
    match value {
        ServerValue::EnrollBound(value) => apply_enroll_bound(aggregate, value),
        ServerValue::Bound(ReceiptReplay::Enrollment(value)) => {
            apply_enroll_bound(aggregate, value);
        }
        ServerValue::AttachBound(value)
        | ServerValue::Bound(ReceiptReplay::CredentialAttach(value)) => {
            apply_attach_bound(aggregate, value);
            aggregate.detach_replay.apply_attach(value);
        }
        ServerValue::UnboundReceipt(ReceiptReplay::CredentialAttach(value)) => {
            apply_unbound_attach_receipt(aggregate, value);
            aggregate.detach_replay.apply_attach(value);
        }
        ServerValue::DetachCommitted(value) => {
            let attach_secret = match aggregate.binding {
                ClientBindingState::Bound { attach_secret, .. }
                | ClientBindingState::Detached { attach_secret, .. } => attach_secret,
                ClientBindingState::Unbound | ClientBindingState::Left { .. } => return,
            };
            aggregate.binding = ClientBindingState::Detached {
                conversation_id: value.conversation_id(),
                participant_id: value.participant_id(),
                generation: value.capability_generation(),
                attach_secret,
            };
            aggregate.detach_replay.apply_detach_committed(value);
        }
        ServerValue::DetachInProgress(value) => {
            aggregate.detach_replay.apply_detach_in_progress(value);
        }
        ServerValue::StaleAuthority(crate::wire::StaleAuthority::Detach(
            crate::wire::DetachStaleAuthority::TerminalizedDetachCell(value),
        )) => {
            aggregate
                .detach_replay
                .apply_terminalized_detach_cell(value);
        }
        ServerValue::LeaveCommitted(value) => {
            aggregate.binding = ClientBindingState::Left {
                conversation_id: value.conversation_id(),
                participant_id: value.participant_id(),
                generation: value.retired_generation(),
            };
            aggregate.detach_replay.apply_leave(value);
        }
        ServerValue::Retired(value) => {
            apply_retired(aggregate, value);
        }
        ServerValue::ParticipantTransportRejected(_)
        | ServerValue::AttemptTokenBodyConflict(_)
        | ServerValue::ConnectionConversationCapacityExceeded(_)
        | ServerValue::ConnectionConversationBindingOccupied(_)
        | ServerValue::ConversationOrderExhausted(_)
        | ServerValue::ParticipantUnknown(_)
        | ServerValue::NoBinding(_)
        | ServerValue::StaleAuthority(_)
        | ServerValue::MarkerClosureCapacityExceeded(_)
        | ServerValue::EnrollmentKnown(_)
        | ServerValue::ReceiptExpired(_)
        | ServerValue::ReceiptCapacityExceeded(_)
        | ServerValue::IdentityCapacityExceeded(_)
        | ServerValue::ObserverBackpressure(_)
        | ServerValue::ConversationSequenceExhausted(_)
        | ServerValue::StaleOrUnknownReceipt(_)
        | ServerValue::MarkerNotDelivered(_)
        | ServerValue::MarkerMismatch(_)
        // Only the ENROLLMENT half stays inert. Its attach twin is adopted
        // above; enrolment's own torn-but-bound recovery is the same shape but
        // a different failure, and closing it here would be closing a failure
        // this lane did not measure.
        | ServerValue::UnboundReceipt(ReceiptReplay::Enrollment(_))
        | ServerValue::AckCommitted(_)
        | ServerValue::AckNoOp(_)
        | ServerValue::AckGap(_)
        | ServerValue::AckRegression(_)
        | ServerValue::MarkerAckCommitted(_)
        | ServerValue::SettlementAnswer(_)
        | ServerValue::RecordCommitted(_)
        | ServerValue::RecordTooLarge(_)
        // A terminal protocol-fault refusal commits nothing and consumes
        // nothing: the server reinstalls the unchanged owner before answering,
        // so no aggregate state moves here either.
        | ServerValue::RecordAdmissionProtocolFault(_)
        | ServerValue::ObserverRecoveryAccepted(_)
        | ServerValue::InvalidObserverEpoch(_)
        | ServerValue::InvalidObserverEpochList(_)
        // Settlement backpressure commits nothing and restores what it
        // consumed, so no aggregate state moves here. The waiting state its
        // retry discipline persists is the SDK's, not the aggregate's
        // (participant contract §0.16 condition 2).
        | ServerValue::MarkerSettlementBackpressure(_)
        | ServerValue::EnrollmentSettlementBackpressure(_)
        // Never correlated to a retirement: it leaves the slot standing.
        | ServerValue::WriteOutcomeUnknown(_) => {}
    }
}

const fn apply_enroll_bound(
    aggregate: &mut ClientParticipantAggregate,
    value: &crate::wire::EnrollBound,
) {
    aggregate.binding = ClientBindingState::Bound {
        conversation_id: value.conversation_id(),
        participant_id: value.participant_id(),
        generation: value.capability_generation(),
        attach_secret: value.attach_secret(),
        binding_epoch: value.origin_binding_epoch(),
    };
}

/// Adopts the rotated credential from an attach receipt whose origin binding is
/// no longer current.
///
/// The server replays this receipt to answer an attempt whose answer was lost,
/// and it carries the two values the client cannot otherwise obtain: the
/// successor capability generation and the newly minted attach secret. Without
/// adopting them the client is stranded one generation behind forever — it can
/// form only an attach the server refuses as `StaleAuthority`, while the attach
/// the server would accept is refused locally as `BindingMismatch` (p0-61
/// residue, task #62).
///
/// The result is [`ClientBindingState::Detached`], never `Bound`. An
/// `UnboundReceipt` states exactly that the receipt no longer names its origin
/// binding — the tear killed the connection that held it — so claiming `Bound`
/// would assert a live binding the server has already released. `Detached`
/// carries the credential forward and is the state a fresh attach may be formed
/// from.
///
/// No identity, liveness or monotonicity guard is written here, and each
/// omission is load-bearing rather than an oversight. [`decide_correlated_inbound`]
/// refuses any value whose expected request the current binding does not accept
/// before application, so a foreign participant's receipt and a receipt arriving
/// after a durable Leave never reach this function; and that same gate forces
/// the retained generation to equal the expected attach's presented generation,
/// which [`AttachBound::ordinary`] guarantees is the exact predecessor of the
/// granted one. A backward adoption is therefore unreachable, not merely
/// guarded. All three are pinned in `p0_62_stranded_handle_tests`.
const fn apply_unbound_attach_receipt(
    aggregate: &mut ClientParticipantAggregate,
    value: &AttachBound,
) {
    aggregate.binding = ClientBindingState::Detached {
        conversation_id: value.conversation_id(),
        participant_id: value.participant_id(),
        generation: value.capability_generation(),
        attach_secret: value.attach_secret(),
    };
}

const fn apply_attach_bound(aggregate: &mut ClientParticipantAggregate, value: &AttachBound) {
    aggregate.binding = ClientBindingState::Bound {
        conversation_id: value.conversation_id(),
        participant_id: value.participant_id(),
        generation: value.capability_generation(),
        attach_secret: value.attach_secret(),
        binding_epoch: value.origin_binding_epoch(),
    };
}

fn apply_retired(aggregate: &mut ClientParticipantAggregate, value: &crate::wire::Retired) {
    let (conversation_id, participant_id, generation) = match value {
        crate::wire::Retired::Enrollment {
            request,
            participant_id,
            retired_generation,
        } => (
            request.conversation_id,
            *participant_id,
            *retired_generation,
        ),
        crate::wire::Retired::Participant {
            request,
            retired_generation,
        } => {
            let (conversation_id, participant_id) = participant_reference_identity(request);
            (conversation_id, participant_id, *retired_generation)
        }
    };
    aggregate.binding = ClientBindingState::Left {
        conversation_id,
        participant_id,
        generation,
    };
    aggregate
        .detach_replay
        .apply_retired(conversation_id, participant_id, generation);
}

const fn participant_reference_identity(
    request: &crate::wire::ParticipantReferenceEnvelope,
) -> (u64, u64) {
    match request {
        crate::wire::ParticipantReferenceEnvelope::CredentialAttach(value) => {
            (value.conversation_id, value.participant_id)
        }
        crate::wire::ParticipantReferenceEnvelope::Detach(value) => {
            (value.conversation_id, value.participant_id)
        }
        crate::wire::ParticipantReferenceEnvelope::ParticipantAck(value) => {
            (value.conversation_id, value.participant_id)
        }
        crate::wire::ParticipantReferenceEnvelope::Leave(value) => {
            (value.conversation_id, value.participant_id)
        }
        crate::wire::ParticipantReferenceEnvelope::MarkerAck(value) => {
            (value.conversation_id, value.participant_id)
        }
        crate::wire::ParticipantReferenceEnvelope::RecordAdmission(value) => {
            (value.conversation_id, value.participant_id)
        }
    }
}