liminal_sdk/remote/

participant.rs

//! Remote participant state, durable client records, and typed transport outcomes.
//!
//! This module owns process mechanics only. Every participant lifecycle and
//! correlation decision is delegated to `liminal-protocol`; the SDK stores the
//! crate aggregate and its sealed one-use authorities without mirroring their
//! rules.

mod recovery;
mod replay_apply;

pub use recovery::{
    RemoteDetachReplayOutcome, RemoteExpectedOperationRecovery, RemoteLostOperationResolution,
    RemoteLostReconnectResolution, RemoteReconnectAttemptOutcome, RemoteReconnectPermitRecovery,
    RemoteReplayApplyOutcome, RemoteTransportLossOutcome,
};

use alloc::sync::Arc;
use core::fmt;
use core::time::Duration;

use liminal_protocol::client::{
    ClientCorrelatedInboundDecision, ClientInboundDecision, ClientInboundRefusalReason,
    ClientOperationRecordDecision, ClientOperationRecordRefusalReason, ClientParticipantAggregate,
    ClientResponseCorrelation, ClientResumeRecord, ClientResumeRecordDecodeError,
    ClientResumeRecordEncodeError, ClientResumeRestoreError, ExpectedOperationFateRefusalReason,
    ExpectedOperationTransportFate, ExpectedParticipantOperation, ReconnectPermitDecision,
    decide_correlated_inbound, decide_inbound, record_expected_operation_fate,
    record_transport_fate,
};
use liminal_protocol::outcome::ReconnectDelayResult;
use liminal_protocol::wire::{
    ClientRequest, DeliverySeq, ParticipantFrame, ServerPush, ServerValue,
};
use spin::Mutex;

use crate::SdkError;

use super::protocol::{ParticipantTransportFrame, RemoteTransport};
use super::{RemoteConfig, ServerAddress};

/// Storage boundary for canonical `LPCR` client resume bytes.
///
/// Implementations must replace the previously committed bytes durably before
/// returning `Ok(())`. The SDK calls this boundary after the protocol crate's
/// commit seal and before releasing executable operation authority.
pub trait ParticipantResumeStore: Send {
    /// Durably replaces the stored canonical client resume record.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Store`] when the bytes were not durably committed.
    fn persist(&mut self, canonical_lpcr: &[u8]) -> Result<(), SdkError>;
}

/// Transport-layer testimony identifying the connection attempt that delivered a frame.
///
/// This is the sealed transport context anticipated by rationale 15 in
/// `LP-CLIENT-GOAL`. It does not alter the wire format or relax the protocol
/// crate's conservative `RecordAdmission` ambiguity.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ParticipantResponseProvenance {
    connection_id: u64,
    attempt_id: u64,
}

impl ParticipantResponseProvenance {
    #[cfg(feature = "std")]
    pub(super) const fn new(connection_id: u64, attempt_id: u64) -> Self {
        Self {
            connection_id,
            attempt_id,
        }
    }

    /// Returns the local identity of the established socket.
    #[must_use]
    pub const fn connection_id(self) -> u64 {
        self.connection_id
    }

    /// Returns the local identity of the real connection attempt.
    #[must_use]
    pub const fn attempt_id(self) -> u64 {
        self.attempt_id
    }
}

/// Failure at the SDK participant state, codec, storage, or transport boundary.
#[derive(Debug, thiserror::Error)]
pub enum RemoteParticipantError {
    /// A prior commit could not be persisted, so no aggregate authority remains reachable.
    ///
    /// `source` is the RETAINED cause: the exact [`SdkError`] the caller's own
    /// store returned at the seam that bricked the handle. It travels on the
    /// error itself, so a caller who merely CAUGHT this value -- through a `?`,
    /// across a channel, out of a join -- can attribute the hold by matching a
    /// typed value rather than parsing rendered text. It is also this error's
    /// [`source`](core::error::Error::source).
    ///
    /// A `None` means the aggregate is unreachable for a reason that did not
    /// originate in the store. That discrimination is why the field is an
    /// `Option` and not an `SdkError`: there is no store failure to name, so
    /// [`source`](core::error::Error::source) reports `None` rather than
    /// inventing a link, and the rendered message drops the durability clause it
    /// would otherwise be asserting without evidence.
    #[error(
        "participant state is unavailable{}",
        source.as_ref().map_or("", |_| " after an unreleased durability failure")
    )]
    StateUnavailable {
        /// The store failure that made the aggregate unreachable, when that is why.
        #[source]
        source: Option<SdkError>,
    },
    /// The protocol crate could not encode the current aggregate as canonical LPCR.
    #[error("client resume record encode failed: {0:?}")]
    ResumeEncode(ClientResumeRecordEncodeError),
    /// Persisted bytes were not a canonical LPCR record.
    #[error("client resume record decode failed: {0:?}")]
    ResumeDecode(ClientResumeRecordDecodeError),
    /// Canonical facts violated a protocol restore invariant.
    #[error("client resume record restore failed: {0:?}")]
    ResumeRestore(ClientResumeRestoreError),
    /// The caller-owned durable store rejected a canonical record.
    ///
    /// The `SdkError` is exposed as this error's [`source`](core::error::Error::source)
    /// so a caller can walk to the typed store failure instead of matching on
    /// rendered text.
    #[error("client resume record persistence failed: {0}")]
    Storage(#[source] SdkError),
    /// The real transport failed outside a typed fate-reporting operation.
    ///
    /// The `SdkError` is exposed as this error's [`source`](core::error::Error::source),
    /// for the same reason as [`Storage`](RemoteParticipantError::Storage).
    #[error("participant transport failed: {0}")]
    Transport(#[source] SdkError),
    /// A client-to-server request appeared on the SDK receive side.
    #[error("participant transport decoded a request in the client receive direction")]
    InvalidInboundDirection,
    /// No live response correlation exists for a replay-specific input.
    #[error("no live participant response authority is held")]
    ResponseAuthorityUnavailable,
}

/// Opaque one-use operation released only after the SDK persisted sealed LPCR bytes.
#[derive(Debug)]
pub struct RemoteParticipantOperation {
    operation: ExpectedParticipantOperation,
    durability: OperationDurability,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum OperationDurability {
    WriteAhead,
    Continuous,
}

/// Result of admitting an outbound request through the crate write-ahead barrier.
#[derive(Debug)]
pub enum RemoteOperationRecordOutcome {
    /// Canonical LPCR bytes were persisted and one operation may now be sent.
    Recorded(RemoteParticipantOperation),
    /// A continuous acknowledgement bypassed the write-ahead slot by crate rule.
    Continuous(RemoteParticipantOperation),
    /// The crate refused the exact request without changing aggregate state.
    Refused {
        /// Exact refused request.
        request: ClientRequest,
        /// Closed protocol refusal reason.
        reason: ClientOperationRecordRefusalReason,
    },
}

/// Typed operation-domain consequence of an established transport loss.
#[derive(Debug, PartialEq, Eq)]
pub enum RemoteOperationTransportFate {
    /// A non-detach operation's response became unavailable.
    Recorded {
        /// Exact terminalized request.
        request: ClientRequest,
    },
    /// The exact detach was returned to parked replay.
    DetachParked,
    /// The crate retained the live correlation unchanged.
    Refused {
        /// Closed refusal reason from the generic operation-fate gate.
        reason: ExpectedOperationFateRefusalReason,
    },
    /// No response authority was outstanding.
    NotOutstanding,
}

/// Typed reconnect permit returned by a crate-authorized fresh event.
#[derive(Debug)]
pub struct RemoteReconnectPermit {
    pub(super) permit: liminal_protocol::client::ReconnectAttemptPermit,
}

/// Event-driven reconnect permit decision; there is no delay or timer arm.
#[derive(Debug)]
pub enum RemoteReconnectPermitOutcome {
    /// The crate minted one one-use permit.
    Permitted {
        /// Opaque permit for one real connection attempt.
        permit: RemoteReconnectPermit,
        /// Legacy-named crate result whose value is event-only.
        result: ReconnectDelayResult,
    },
    /// Existing authority was retained.
    Refused {
        /// Closed crate refusal reason.
        reason: liminal_protocol::client::ReconnectPermitRefusalReason,
        /// Event-only crate result.
        result: ReconnectDelayResult,
    },
}

/// Result of sending an operation on the participant transport.
#[derive(Debug)]
pub enum RemoteParticipantSendOutcome {
    /// The request bytes were written on this connection attempt.
    Sent {
        /// Sealed transport context for a later response.
        provenance: ParticipantResponseProvenance,
    },
    /// The write failed and both operation and reconnect fates were delegated.
    TransportLost {
        /// Concrete socket failure.
        error: SdkError,
        /// Crate-owned operation-fate result.
        operation_fate: RemoteOperationTransportFate,
        /// Crate-owned reconnect permit result.
        reconnect: RemoteReconnectPermitOutcome,
    },
}

/// Default quiet window for [`RemoteParticipantHandle::try_receive`].
///
/// Deliberately ONE steady-state transport receive window rather than a new
/// number of its own: that is the grain both transports already poll on, and it
/// is the shape consumers' drain loops were written against before a total
/// response deadline was layered above it. A pump that reports quiet after one
/// closed window is the behaviour they expect; the deadline above it is the
/// reply-owed protection they were never asking for.
pub const PARTICIPANT_PUMP_WINDOW: Duration = super::framing::IO_TIMEOUT;

/// Typed result of one decoded participant frame on the real receive path.
#[derive(Debug)]
pub enum RemoteParticipantInbound {
    /// The protocol crate correlated and applied a semantic response.
    Applied {
        /// Exact applied server value.
        value: ServerValue,
        /// Connection/attempt that delivered it.
        provenance: ParticipantResponseProvenance,
    },
    /// The protocol crate retained the response and aggregate unchanged.
    Refused {
        /// Exact refused server value.
        value: ServerValue,
        /// Closed crate refusal reason, including conservative ambiguity.
        reason: ClientInboundRefusalReason,
        /// Connection/attempt that delivered it.
        provenance: ParticipantResponseProvenance,
    },
    /// Server push decoded in the client direction; no correlation rule applies.
    Push {
        /// Exact pushed value.
        value: ServerPush,
        /// Connection/attempt that delivered it.
        provenance: ParticipantResponseProvenance,
    },
}

impl RemoteParticipantInbound {
    /// The record sequence the server assigned an admitted record, when this
    /// inbound is an APPLIED [`ServerValue::RecordCommitted`].
    ///
    /// This is the answer to a `RecordAdmission`, read off the exact wire value
    /// the protocol crate applied. It saves every caller destructuring the wire
    /// enum to reach the one number a record admission is asked for, without
    /// removing that value: [`Applied`](Self::Applied) still carries the whole
    /// `ServerValue`, so this is purely additive.
    ///
    /// `None` for everything else, and that includes a `RecordCommitted` the
    /// crate REFUSED. A refused commit carries a sequence on the wire while
    /// leaving the aggregate and its correlation untouched -- returning it here
    /// would report a commitment the crate deliberately declined to make. It is
    /// also `None` for a [`Push`](Self::Push), which is a delivery rather than
    /// a correlated response.
    #[must_use]
    pub const fn committed_delivery_seq(&self) -> Option<DeliverySeq> {
        match self {
            Self::Applied {
                value: ServerValue::RecordCommitted(committed),
                ..
            } => Some(committed.delivery_seq()),
            Self::Applied { .. } | Self::Refused { .. } | Self::Push { .. } => None,
        }
    }
}

pub(super) struct RemoteParticipantState<S> {
    pub(super) aggregate: Option<ClientParticipantAggregate>,
    pub(super) correlation: Option<ClientResponseCorrelation>,
    pub(super) reconnect_attempt: Option<liminal_protocol::client::ReconnectInProgressAttempt>,
    pub(super) store: S,
    /// The store failure that made the aggregate unreachable, if that is why.
    ///
    /// The failure itself is returned to whoever made the failing call and to
    /// nobody else. This retains it so the question "why is this handle dead"
    /// has a typed answer for every later caller: `take_aggregate` reads it back
    /// out of here onto the
    /// [`StateUnavailable`](RemoteParticipantError::StateUnavailable) it reports,
    /// and [`unavailability_cause`](RemoteParticipantHandle::unavailability_cause)
    /// answers it without a failing call.
    pub(super) unavailable: Option<SdkError>,
}

impl<S> RemoteParticipantState<S> {
    /// Records a store failure as the reason the aggregate is about to become
    /// unreachable, and names that failure for the caller.
    ///
    /// Every seam that drops the aggregate on a store failure goes through here,
    /// so the retained cause cannot drift from the returned one: they are the
    /// same value.
    pub(super) fn brick(&mut self, error: SdkError) -> RemoteParticipantError {
        self.unavailable = Some(error.clone());
        RemoteParticipantError::Storage(error)
    }
}

/// Remote participant entrypoint backed by protocol-crate state and canonical LPCR storage.
///
/// Records are deliberately not promised as generally successful: the reduced-B1
/// server surface fails fully authorized `RecordAdmission` and `Leave` closed until
/// live claim-frontier acquisition lands (`docs/design/LP-GAP-CLOSURE-GOAL.md:145`).
pub struct RemoteParticipantHandle<S> {
    pub(super) server_address: ServerAddress,
    pub(super) transport: Arc<dyn RemoteTransport>,
    pub(super) state: Mutex<RemoteParticipantState<S>>,
}

impl<S> fmt::Debug for RemoteParticipantHandle<S> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("RemoteParticipantHandle")
            .field("server_address", &self.server_address)
            .finish_non_exhaustive()
    }
}

impl<S: ParticipantResumeStore> RemoteParticipantHandle<S> {
    /// Creates and durably checkpoints a fresh unbound participant.
    ///
    /// # Errors
    ///
    /// Returns a typed encode or storage error before the handle is exposed.
    pub fn new(config: &RemoteConfig, store: S) -> Result<Self, RemoteParticipantError> {
        Self::from_aggregate(config, store, ClientParticipantAggregate::new())
    }

    /// Decodes, validates, restores, and durably records crash testimony before exposure.
    ///
    /// # Errors
    ///
    /// Returns typed LPCR decode/restore, encode, or storage errors.
    pub fn restore(
        config: &RemoteConfig,
        store: S,
        canonical_lpcr: &[u8],
    ) -> Result<Self, RemoteParticipantError> {
        let record = ClientResumeRecord::decode_canonical(canonical_lpcr)
            .map_err(RemoteParticipantError::ResumeDecode)?;
        let aggregate = record
            .restore()
            .map_err(RemoteParticipantError::ResumeRestore)?;
        Self::from_aggregate(config, store, aggregate)
    }

    fn from_aggregate(
        config: &RemoteConfig,
        mut store: S,
        aggregate: ClientParticipantAggregate,
    ) -> Result<Self, RemoteParticipantError> {
        persist(&mut store, &aggregate)?;
        Ok(Self {
            server_address: config.server_address.clone(),
            transport: Arc::clone(&config.transport),
            state: Mutex::new(RemoteParticipantState {
                aggregate: Some(aggregate),
                correlation: None,
                reconnect_attempt: None,
                store,
                unavailable: None,
            }),
        })
    }

    /// Returns the store failure that made this handle unavailable, if that is
    /// why it is unavailable.
    ///
    /// A `Some` names the exact [`SdkError`] the caller's own store returned, so
    /// a hold can be attributed by matching a typed value. A `None` means the
    /// handle is either live or unavailable for a reason that did not originate
    /// in the store — which is itself the discrimination this exists to provide.
    ///
    /// The same retained value is carried on
    /// [`StateUnavailable`](RemoteParticipantError::StateUnavailable) and is that
    /// error's [`source`](core::error::Error::source), so a caller who CAUGHT
    /// that error already holds the cause and does not need this. This accessor
    /// answers the other question — one asked of a handle you still HOLD, rather
    /// than of an error you caught — and it answers it without making a call
    /// that must fail in order to find out. Its `None` therefore covers one case
    /// the variant's `None` cannot: a handle that is simply still live.
    #[must_use]
    pub fn unavailability_cause(&self) -> Option<SdkError> {
        self.state.lock().unavailable.clone()
    }

    /// Runs `record_operation -> commit -> LPCR persist -> into_parts` exactly.
    ///
    /// # Errors
    ///
    /// Returns typed resume encoding or storage failures. A failed post-commit
    /// persistence leaves the handle unavailable and releases no authority.
    pub fn record_operation(
        &self,
        request: ClientRequest,
    ) -> Result<RemoteOperationRecordOutcome, RemoteParticipantError> {
        let mut state = self.state.lock();
        let aggregate = take_aggregate(&mut state)?;
        match liminal_protocol::client::record_operation(aggregate, request) {
            ClientOperationRecordDecision::Pending(pending) => {
                let commit = pending.commit();
                let record = commit
                    .resume_record()
                    .map_err(RemoteParticipantError::ResumeEncode)?;
                if let Err(error) = state.store.persist(&record.encode_canonical()) {
                    return Err(state.brick(error));
                }
                let (aggregate, operation) = commit.into_parts();
                state.aggregate = Some(aggregate);
                Ok(RemoteOperationRecordOutcome::Recorded(
                    RemoteParticipantOperation {
                        operation,
                        durability: OperationDurability::WriteAhead,
                    },
                ))
            }
            ClientOperationRecordDecision::Continuous(continuous) => {
                let (aggregate, operation) = continuous.into_parts();
                state.aggregate = Some(aggregate);
                Ok(RemoteOperationRecordOutcome::Continuous(
                    RemoteParticipantOperation {
                        operation,
                        durability: OperationDurability::Continuous,
                    },
                ))
            }
            ClientOperationRecordDecision::Refused(refusal) => {
                let reason = refusal.reason();
                let (aggregate, request) = refusal.into_parts();
                state.aggregate = Some(aggregate);
                Ok(RemoteOperationRecordOutcome::Refused { request, reason })
            }
        }
    }

    /// Persists issued state, writes the exact operation, and retains correlation.
    ///
    /// # Errors
    ///
    /// Returns typed state, LPCR, or storage failures. Transport failures are
    /// returned as a typed outcome after crate fate delegation.
    pub fn send_operation(
        &self,
        operation: RemoteParticipantOperation,
    ) -> Result<RemoteParticipantSendOutcome, RemoteParticipantError> {
        let mut state = self.state.lock();
        let mut aggregate = take_aggregate(&mut state)?;
        if operation.durability == OperationDurability::WriteAhead {
            aggregate = persist_retaining(&mut state, aggregate)?;
        }
        let (request, correlation) = operation.operation.into_request();
        match self
            .transport
            .send_participant(&self.server_address, &request)
        {
            Ok(provenance) => {
                if operation.durability == OperationDurability::WriteAhead {
                    state.correlation = Some(correlation);
                }
                state.aggregate = Some(aggregate);
                Ok(RemoteParticipantSendOutcome::Sent { provenance })
            }
            Err(error) => {
                let operation_fate = if operation.durability == OperationDurability::WriteAhead {
                    record_operation_transport_fate(&mut state, aggregate, correlation)
                } else {
                    state.aggregate = Some(aggregate);
                    RemoteOperationTransportFate::NotOutstanding
                };
                let reconnect = record_connection_fate(&mut state)?;
                Ok(RemoteParticipantSendOutcome::TransportLost {
                    error,
                    operation_fate,
                    reconnect,
                })
            }
        }
    }

    /// Receives one real participant frame and delegates every `ServerValue` to the crate.
    ///
    /// # The contract, and why it is this one
    ///
    /// THIS IS THE REPLY-OWED DOOR. It blocks for up to the transport's full
    /// response deadline (60 s), because the caller it is written for has just
    /// sent a request and is waiting for the correlated answer — and there, a
    /// quiet connection means a slow server, not a dead one. Ending that wait
    /// early is the 2026-08-10 outage's client-side mechanism, so this method
    /// keeps the deadline unchanged and unconditionally.
    ///
    /// A consumer PUMPING an idle connection — looping to collect whatever the
    /// server pushes next, where silence is a normal state rather than a fault
    /// — must use [`receive_within`](Self::receive_within) or
    /// [`try_receive`](Self::try_receive) instead. That is not a preference: a
    /// drain loop built on this method waits out the full deadline on every
    /// quiet read, which is how a 30 s boot gate blows on a healthy server.
    ///
    /// The split is by CALLER INTENT and cannot be anything else. At a clean
    /// frame boundary with an empty buffer, a pump read and an outage-shaped
    /// reply-owed read are byte-for-byte identical states; only the caller
    /// knows which one it is making, so only the caller's choice of method can
    /// carry it. Inferring it from buffered bytes, or from whether an operation
    /// is outstanding, would silently shorten the deadline for some class of
    /// genuinely reply-owed read and re-open the outage for it.
    ///
    /// Pushed deliveries are at-least-once: the same
    /// `(conversation_id, delivery_seq)` may arrive more than once on one
    /// healthy connection, byte-identical each time — deduplicate on the pair
    /// (participant contract R-C3, amendment A3).
    ///
    /// # Errors
    ///
    /// Returns transport, direction, LPCR encoding, or storage failures.
    pub fn receive(&self) -> Result<RemoteParticipantInbound, RemoteParticipantError> {
        let frame = self
            .transport
            .receive_participant(&self.server_address)
            .map_err(RemoteParticipantError::Transport)?;
        self.classify_inbound(frame)
    }

    /// Receives one participant frame if one arrives within `budget`, reporting
    /// a quiet connection as `Ok(None)` instead of an error.
    ///
    /// THE PUMP DOOR — the lawful read for a consumer that is owed nothing.
    /// `Ok(None)` means "no frame within this window", which is a normal state
    /// on a healthy connection and never a fault; a real transport failure
    /// still returns `Err`, and a quiet window never surfaces a raw errno.
    ///
    /// `budget` is the CALLER'S bound and is spent across as many transport
    /// read windows as it takes. It never shortens anything else: a caller that
    /// uses this method to await a correlated answer simply names its own
    /// deadline, and passing one at or above the transport's 60 s response
    /// deadline reproduces [`receive`](Self::receive)'s patience with a typed
    /// silence at the end instead of an error.
    ///
    /// `Duration::ZERO` polls only what has already decoded, without arming a
    /// read. Bytes of a partly-arrived frame stay buffered across an
    /// `Ok(None)`, so a frame that was mid-flight when the budget expired is
    /// never lost — the next call resumes on it.
    ///
    /// # Errors
    ///
    /// Returns transport, direction, LPCR encoding, or storage failures. A
    /// quiet window is NOT one of them.
    pub fn receive_within(
        &self,
        budget: Duration,
    ) -> Result<Option<RemoteParticipantInbound>, RemoteParticipantError> {
        let Some(frame) = self
            .transport
            .receive_participant_within(&self.server_address, budget)
            .map_err(RemoteParticipantError::Transport)?
        else {
            return Ok(None);
        };
        self.classify_inbound(frame).map(Some)
    }

    /// One [`PARTICIPANT_PUMP_WINDOW`] of patience, then `Ok(None)`.
    ///
    /// The drain-loop convenience over [`receive_within`](Self::receive_within):
    /// a consumer that wants "give me the next frame, or tell me the connection
    /// is quiet" without choosing a number. Loop it until it answers `Ok(None)`
    /// and the backlog is drained.
    ///
    /// # Errors
    ///
    /// As [`receive_within`](Self::receive_within).
    pub fn try_receive(&self) -> Result<Option<RemoteParticipantInbound>, RemoteParticipantError> {
        self.receive_within(PARTICIPANT_PUMP_WINDOW)
    }

    /// Routes one decoded transport frame into the crate's inbound decisions.
    ///
    /// Shared by [`receive`](Self::receive) and
    /// [`receive_within`](Self::receive_within) so the two doors differ ONLY in
    /// how long they wait for a frame. Every correlation, application, and
    /// refusal rule below is reached identically by both — a pump read that
    /// does find a frame applies it exactly as a reply-owed read would.
    fn classify_inbound(
        &self,
        frame: ParticipantTransportFrame,
    ) -> Result<RemoteParticipantInbound, RemoteParticipantError> {
        let ParticipantTransportFrame { frame, provenance } = frame;
        match frame {
            ParticipantFrame::ServerPush(value) => {
                Ok(RemoteParticipantInbound::Push { value, provenance })
            }
            ParticipantFrame::ClientRequest(_) => {
                Err(RemoteParticipantError::InvalidInboundDirection)
            }
            ParticipantFrame::ServerValue(value) => self.apply_inbound(value, provenance),
        }
    }

    fn apply_inbound(
        &self,
        value: ServerValue,
        provenance: ParticipantResponseProvenance,
    ) -> Result<RemoteParticipantInbound, RemoteParticipantError> {
        let mut state = self.state.lock();
        let aggregate = take_aggregate(&mut state)?;
        if let Some(correlation) = state.correlation.take() {
            match decide_correlated_inbound(aggregate, value, correlation) {
                ClientCorrelatedInboundDecision::Applied(applied) => {
                    let (aggregate, value) = applied.into_parts();
                    let aggregate = persist_retaining(&mut state, aggregate)?;
                    state.aggregate = Some(aggregate);
                    Ok(RemoteParticipantInbound::Applied { value, provenance })
                }
                ClientCorrelatedInboundDecision::Refused(refusal) => {
                    let reason = refusal.reason();
                    let (aggregate, value, correlation) = refusal.into_parts();
                    state.aggregate = Some(aggregate);
                    state.correlation = Some(correlation);
                    Ok(RemoteParticipantInbound::Refused {
                        value,
                        reason,
                        provenance,
                    })
                }
            }
        } else {
            match decide_inbound(aggregate, value) {
                ClientInboundDecision::Applied(applied) => {
                    let (aggregate, value) = applied.into_parts();
                    let aggregate = persist_retaining(&mut state, aggregate)?;
                    state.aggregate = Some(aggregate);
                    Ok(RemoteParticipantInbound::Applied { value, provenance })
                }
                ClientInboundDecision::Refused(refusal) => {
                    let reason = refusal.reason();
                    let (aggregate, value) = refusal.into_parts();
                    state.aggregate = Some(aggregate);
                    Ok(RemoteParticipantInbound::Refused {
                        value,
                        reason,
                        provenance,
                    })
                }
            }
        }
    }
}

pub(super) fn take_aggregate<S>(
    state: &mut RemoteParticipantState<S>,
) -> Result<ClientParticipantAggregate, RemoteParticipantError> {
    let aggregate = state.aggregate.take();
    aggregate.ok_or_else(|| RemoteParticipantError::StateUnavailable {
        source: state.unavailable.clone(),
    })
}

pub(super) fn persist<S: ParticipantResumeStore>(
    store: &mut S,
    aggregate: &ClientParticipantAggregate,
) -> Result<(), RemoteParticipantError> {
    let record = aggregate
        .resume_record()
        .map_err(RemoteParticipantError::ResumeEncode)?;
    store
        .persist(&record.encode_canonical())
        .map_err(RemoteParticipantError::Storage)
}

/// Persists `aggregate`, returning it to the caller on success and RE-SEATING it
/// in `state` when encoding refused.
///
/// The two failure modes are not alike, and the difference is durability
/// ambiguity:
///
/// * [`RemoteParticipantError::Storage`] means the store was asked to commit
///   bytes and did not say it succeeded. Whether those bytes landed is unknown,
///   so no further authority may be released from this aggregate. The handle
///   is deliberately left bricked and the next call reports
///   [`StateUnavailable`](RemoteParticipantError::StateUnavailable) -- the
///   contract documented on that variant.
/// * [`RemoteParticipantError::ResumeEncode`] means `resume_record`, a pure
///   function of the aggregate, refused. Nothing was written and no authority
///   was released, so there is nothing ambiguous to protect. Dropping the
///   aggregate here converts a typed, catchable refusal into a permanently dead
///   participant.
///
/// The re-seat lives in this function rather than at each seam on purpose: the
/// SDK has ten `take_aggregate` -> persist -> re-seat sites, the `?` skips the
/// re-seat at every one of them, and a new seam would inherit the same bug.
/// Here it cannot be forgotten.
///
/// A caller that still needs the aggregate takes it back from the `Ok`; a
/// caller that does not re-seats it itself. On the `ResumeEncode` path the
/// aggregate is already seated, so callers must not seat it again.
pub(super) fn persist_retaining<S: ParticipantResumeStore>(
    state: &mut RemoteParticipantState<S>,
    aggregate: ClientParticipantAggregate,
) -> Result<ClientParticipantAggregate, RemoteParticipantError> {
    match aggregate.resume_record() {
        Ok(record) => match state.store.persist(&record.encode_canonical()) {
            Ok(()) => Ok(aggregate),
            // The aggregate is deliberately NOT re-seated: whether the bytes
            // landed is unknown, so no further authority may be released. The
            // cause is retained on the way past, because it is about to be the
            // only thing that could ever explain the `StateUnavailable` every
            // later call will report.
            Err(error) => Err(state.brick(error)),
        },
        Err(error) => {
            state.aggregate = Some(aggregate);
            Err(RemoteParticipantError::ResumeEncode(error))
        }
    }
}

fn record_operation_transport_fate<S: ParticipantResumeStore>(
    state: &mut RemoteParticipantState<S>,
    aggregate: ClientParticipantAggregate,
    correlation: ClientResponseCorrelation,
) -> RemoteOperationTransportFate {
    match record_expected_operation_fate(
        aggregate,
        correlation,
        ExpectedOperationTransportFate::ResponseUnavailable,
    ) {
        liminal_protocol::client::ExpectedOperationFateDecision::Recorded {
            aggregate,
            request,
            ..
        } => {
            state.aggregate = Some(aggregate);
            RemoteOperationTransportFate::Recorded { request }
        }
        liminal_protocol::client::ExpectedOperationFateDecision::Refused {
            aggregate,
            correlation,
            reason: ExpectedOperationFateRefusalReason::DetachUsesReplayFate,
            ..
        } => match liminal_protocol::client::transport_fate(
            aggregate,
            correlation,
            liminal_protocol::client::DetachTransportFate::ResponseUnavailable,
        ) {
            liminal_protocol::client::DetachTransportFateDecision::Parked(applied) => {
                state.aggregate = Some(applied.into_aggregate());
                RemoteOperationTransportFate::DetachParked
            }
            liminal_protocol::client::DetachTransportFateDecision::Refused(refusal) => {
                let (aggregate, (correlation, _)) = refusal.into_parts();
                state.aggregate = Some(aggregate);
                state.correlation = Some(correlation);
                RemoteOperationTransportFate::Refused {
                    reason: ExpectedOperationFateRefusalReason::DetachUsesReplayFate,
                }
            }
        },
        liminal_protocol::client::ExpectedOperationFateDecision::Refused {
            aggregate,
            correlation,
            reason,
            ..
        } => {
            state.aggregate = Some(aggregate);
            state.correlation = Some(correlation);
            RemoteOperationTransportFate::Refused { reason }
        }
    }
}

pub(super) fn record_connection_fate<S: ParticipantResumeStore>(
    state: &mut RemoteParticipantState<S>,
) -> Result<RemoteReconnectPermitOutcome, RemoteParticipantError> {
    let aggregate = take_aggregate(state)?;
    let (aggregate, outcome) = match record_transport_fate(
        aggregate,
        liminal_protocol::client::EstablishedConnectionTransportFate::Lost,
    ) {
        ReconnectPermitDecision::Permitted {
            aggregate,
            permit,
            result,
        } => (
            aggregate,
            RemoteReconnectPermitOutcome::Permitted {
                permit: RemoteReconnectPermit { permit },
                result,
            },
        ),
        ReconnectPermitDecision::Refused(refusal) => {
            let reason = refusal.reason();
            let result = refusal.result();
            let (aggregate, _) = refusal.into_parts();
            (
                aggregate,
                RemoteReconnectPermitOutcome::Refused { reason, result },
            )
        }
    };
    let aggregate = persist_retaining(state, aggregate)?;
    state.aggregate = Some(aggregate);
    Ok(outcome)
}

#[cfg(test)]
mod tests;