liminal_protocol/client/
inbound.rs

1use super::{
2    ClientBindingState, ClientParticipantAggregate, ClientResponseCorrelation, correlation,
3};
4use crate::wire::{AttachBound, ReceiptReplay, ServerValue};
5
6/// Closed refusal classes for inbound semantic values.
7#[derive(Clone, Copy, Debug, PartialEq, Eq)]
8pub enum ClientInboundRefusalReason {
9    /// A durable Leave already terminalized the local participant.
10    AlreadyDead,
11    /// The value names another operation or participant identity.
12    ForeignResponse,
13    /// The value is absent an expectation or belongs to an older request.
14    DelayedResponse,
15    /// Wire identity is insufficient to assign this value to one expected operation.
16    AmbiguousResponse,
17    /// An expected-operation response was presented without its one-use send correlation.
18    MissingResponseAuthority,
19    /// A restore already testified the issued send authority destroyed; only
20    /// the pending testimony can resolve the operation (r2, 2026-07-18).
21    LostAuthorityPending,
22}
23
24/// Applied inbound value and resulting aggregate.
25#[derive(Debug, PartialEq, Eq)]
26pub struct ClientInboundApplied {
27    aggregate: ClientParticipantAggregate,
28    value: ServerValue,
29}
30
31impl ClientInboundApplied {
32    /// Releases the resulting aggregate and exact applied value.
33    #[must_use]
34    pub fn into_parts(self) -> (ClientParticipantAggregate, ServerValue) {
35        (self.aggregate, self.value)
36    }
37}
38
39/// Refused inbound value paired with the unchanged aggregate.
40#[derive(Debug, PartialEq, Eq)]
41pub struct ClientInboundRefusal {
42    aggregate: ClientParticipantAggregate,
43    value: ServerValue,
44    reason: ClientInboundRefusalReason,
45}
46
47impl ClientInboundRefusal {
48    /// Returns the closed refusal reason.
49    #[must_use]
50    pub const fn reason(&self) -> ClientInboundRefusalReason {
51        self.reason
52    }
53
54    /// Releases the unchanged aggregate and exact refused value.
55    #[must_use]
56    pub fn into_parts(self) -> (ClientParticipantAggregate, ServerValue) {
57        (self.aggregate, self.value)
58    }
59}
60
61/// Exhaustive inbound correlation decision.
62#[derive(Debug, PartialEq, Eq)]
63pub enum ClientInboundDecision {
64    /// The crate correlated and applied the typed value.
65    Applied(ClientInboundApplied),
66    /// The crate retained both authority and value unchanged.
67    Refused(ClientInboundRefusal),
68}
69
70/// Refused body-omitting response with the exact local correlation retained.
71#[derive(Debug, PartialEq, Eq)]
72pub struct ClientCorrelatedInboundRefusal {
73    aggregate: ClientParticipantAggregate,
74    value: ServerValue,
75    correlation: ClientResponseCorrelation,
76    reason: ClientInboundRefusalReason,
77}
78
79impl ClientCorrelatedInboundRefusal {
80    /// Returns the closed refusal reason.
81    #[must_use]
82    pub const fn reason(&self) -> ClientInboundRefusalReason {
83        self.reason
84    }
85
86    /// Releases every unchanged input, including the non-cloneable correlation.
87    #[must_use]
88    pub fn into_parts(
89        self,
90    ) -> (
91        ClientParticipantAggregate,
92        ServerValue,
93        ClientResponseCorrelation,
94    ) {
95        (self.aggregate, self.value, self.correlation)
96    }
97}
98
99/// Inbound decision for response classes whose wire envelopes omit request identity.
100#[derive(Debug, PartialEq, Eq)]
101pub enum ClientCorrelatedInboundDecision {
102    /// The exact local operation authorization and wire envelope both matched.
103    Applied(ClientInboundApplied),
104    /// Aggregate, value, and correlation were retained unchanged.
105    Refused(ClientCorrelatedInboundRefusal),
106}
107
108/// Correlates and applies one server value inside the client aggregate.
109#[must_use]
110pub fn decide_inbound(
111    aggregate: ClientParticipantAggregate,
112    value: ServerValue,
113) -> ClientInboundDecision {
114    decide_inbound_inner(aggregate, value, false)
115}
116
117/// Attempts correlation while retaining the process-local handle on refusal.
118#[must_use]
119pub fn decide_correlated_inbound(
120    aggregate: ClientParticipantAggregate,
121    value: ServerValue,
122    correlation: ClientResponseCorrelation,
123) -> ClientCorrelatedInboundDecision {
124    let current_authority = aggregate.expected.as_ref().is_some_and(|expected| {
125        expected.issued && expected.authorization == correlation.authorization
126    });
127    if !current_authority {
128        return ClientCorrelatedInboundDecision::Refused(ClientCorrelatedInboundRefusal {
129            aggregate,
130            value,
131            correlation,
132            reason: ClientInboundRefusalReason::DelayedResponse,
133        });
134    }
135    match decide_inbound_inner(aggregate, value, true) {
136        ClientInboundDecision::Applied(applied) => {
137            ClientCorrelatedInboundDecision::Applied(applied)
138        }
139        ClientInboundDecision::Refused(refusal) => {
140            let reason = refusal.reason();
141            let (aggregate, value) = refusal.into_parts();
142            ClientCorrelatedInboundDecision::Refused(ClientCorrelatedInboundRefusal {
143                aggregate,
144                value,
145                correlation,
146                reason,
147            })
148        }
149    }
150}
151
152fn decide_inbound_inner(
153    mut aggregate: ClientParticipantAggregate,
154    value: ServerValue,
155    has_response_authority: bool,
156) -> ClientInboundDecision {
157    if aggregate.binding.is_left() {
158        return inbound_refusal(aggregate, value, ClientInboundRefusalReason::AlreadyDead);
159    }
160
161    if let Some(request) = correlation::participant_ack_request(&value) {
162        if aggregate.binding.matches_ack(request) {
163            return ClientInboundDecision::Applied(ClientInboundApplied { aggregate, value });
164        }
165        return inbound_refusal(
166            aggregate,
167            value,
168            ClientInboundRefusalReason::ForeignResponse,
169        );
170    }
171
172    if matches!(value, ServerValue::ParticipantTransportRejected(_)) {
173        return ClientInboundDecision::Applied(ClientInboundApplied { aggregate, value });
174    }
175
176    let Some(expected) = aggregate.expected.as_ref() else {
177        return inbound_refusal(
178            aggregate,
179            value,
180            ClientInboundRefusalReason::DelayedResponse,
181        );
182    };
183
184    if expected.lost.is_some() {
185        return inbound_refusal(
186            aggregate,
187            value,
188            ClientInboundRefusalReason::LostAuthorityPending,
189        );
190    }
191
192    if !has_response_authority {
193        return inbound_refusal(
194            aggregate,
195            value,
196            ClientInboundRefusalReason::MissingResponseAuthority,
197        );
198    }
199
200    if !aggregate.binding.accepts_request(&expected.request) {
201        return inbound_refusal(
202            aggregate,
203            value,
204            ClientInboundRefusalReason::ForeignResponse,
205        );
206    }
207
208    if !correlation::matches_request(&value, &expected.request) {
209        let same_request_class = value.originating_request()
210            == Some(expected.request.discriminant())
211            || matches!(
212                (&value, &expected.request),
213                (
214                    ServerValue::ObserverRecoveryAccepted(_)
215                        | ServerValue::InvalidObserverEpoch(_)
216                        | ServerValue::InvalidObserverEpochList(_),
217                    crate::wire::ClientRequest::ObserverRecovery(_)
218                )
219            );
220        let same_identity = correlation::same_identity(&value, &expected.request);
221        let reason = if same_request_class && same_identity {
222            if matches!(
223                expected.request,
224                crate::wire::ClientRequest::RecordAdmission(_)
225            ) {
226                ClientInboundRefusalReason::AmbiguousResponse
227            } else {
228                ClientInboundRefusalReason::DelayedResponse
229            }
230        } else {
231            ClientInboundRefusalReason::ForeignResponse
232        };
233        return inbound_refusal(aggregate, value, reason);
234    }
235
236    aggregate.expected = None;
237    apply_correlated_value(&mut aggregate, &value);
238    ClientInboundDecision::Applied(ClientInboundApplied { aggregate, value })
239}
240
241const fn inbound_refusal(
242    aggregate: ClientParticipantAggregate,
243    value: ServerValue,
244    reason: ClientInboundRefusalReason,
245) -> ClientInboundDecision {
246    ClientInboundDecision::Refused(ClientInboundRefusal {
247        aggregate,
248        value,
249        reason,
250    })
251}
252
253fn apply_correlated_value(aggregate: &mut ClientParticipantAggregate, value: &ServerValue) {
254    match value {
255        ServerValue::EnrollBound(value) => apply_enroll_bound(aggregate, value),
256        ServerValue::Bound(ReceiptReplay::Enrollment(value)) => {
257            apply_enroll_bound(aggregate, value);
258        }
259        ServerValue::AttachBound(value)
260        | ServerValue::Bound(ReceiptReplay::CredentialAttach(value)) => {
261            apply_attach_bound(aggregate, value);
262            aggregate.detach_replay.apply_attach(value);
263        }
264        ServerValue::DetachCommitted(value) => {
265            let attach_secret = match aggregate.binding {
266                ClientBindingState::Bound { attach_secret, .. }
267                | ClientBindingState::Detached { attach_secret, .. } => attach_secret,
268                ClientBindingState::Unbound | ClientBindingState::Left { .. } => return,
269            };
270            aggregate.binding = ClientBindingState::Detached {
271                conversation_id: value.conversation_id(),
272                participant_id: value.participant_id(),
273                generation: value.capability_generation(),
274                attach_secret,
275            };
276            aggregate.detach_replay.apply_detach_committed(value);
277        }
278        ServerValue::DetachInProgress(value) => {
279            aggregate.detach_replay.apply_detach_in_progress(value);
280        }
281        ServerValue::StaleAuthority(crate::wire::StaleAuthority::Detach(
282            crate::wire::DetachStaleAuthority::TerminalizedDetachCell(value),
283        )) => {
284            aggregate
285                .detach_replay
286                .apply_terminalized_detach_cell(value);
287        }
288        ServerValue::LeaveCommitted(value) => {
289            aggregate.binding = ClientBindingState::Left {
290                conversation_id: value.conversation_id(),
291                participant_id: value.participant_id(),
292                generation: value.retired_generation(),
293            };
294            aggregate.detach_replay.apply_leave(value);
295        }
296        ServerValue::Retired(value) => {
297            apply_retired(aggregate, value);
298        }
299        ServerValue::ParticipantTransportRejected(_)
300        | ServerValue::AttemptTokenBodyConflict(_)
301        | ServerValue::ConnectionConversationCapacityExceeded(_)
302        | ServerValue::ConnectionConversationBindingOccupied(_)
303        | ServerValue::ConversationOrderExhausted(_)
304        | ServerValue::ParticipantUnknown(_)
305        | ServerValue::NoBinding(_)
306        | ServerValue::StaleAuthority(_)
307        | ServerValue::MarkerClosureCapacityExceeded(_)
308        | ServerValue::EnrollmentKnown(_)
309        | ServerValue::ReceiptExpired(_)
310        | ServerValue::ReceiptCapacityExceeded(_)
311        | ServerValue::IdentityCapacityExceeded(_)
312        | ServerValue::ObserverBackpressure(_)
313        | ServerValue::ConversationSequenceExhausted(_)
314        | ServerValue::StaleOrUnknownReceipt(_)
315        | ServerValue::MarkerNotDelivered(_)
316        | ServerValue::MarkerMismatch(_)
317        | ServerValue::UnboundReceipt(_)
318        | ServerValue::AckCommitted(_)
319        | ServerValue::AckNoOp(_)
320        | ServerValue::AckGap(_)
321        | ServerValue::AckRegression(_)
322        | ServerValue::MarkerAckCommitted(_)
323        | ServerValue::RecordCommitted(_)
324        | ServerValue::RecordTooLarge(_)
325        | ServerValue::ObserverRecoveryAccepted(_)
326        | ServerValue::InvalidObserverEpoch(_)
327        | ServerValue::InvalidObserverEpochList(_) => {}
328    }
329}
330
331const fn apply_enroll_bound(
332    aggregate: &mut ClientParticipantAggregate,
333    value: &crate::wire::EnrollBound,
334) {
335    aggregate.binding = ClientBindingState::Bound {
336        conversation_id: value.conversation_id(),
337        participant_id: value.participant_id(),
338        generation: value.capability_generation(),
339        attach_secret: value.attach_secret(),
340        binding_epoch: value.origin_binding_epoch(),
341    };
342}
343
344const fn apply_attach_bound(aggregate: &mut ClientParticipantAggregate, value: &AttachBound) {
345    aggregate.binding = ClientBindingState::Bound {
346        conversation_id: value.conversation_id(),
347        participant_id: value.participant_id(),
348        generation: value.capability_generation(),
349        attach_secret: value.attach_secret(),
350        binding_epoch: value.origin_binding_epoch(),
351    };
352}
353
354fn apply_retired(aggregate: &mut ClientParticipantAggregate, value: &crate::wire::Retired) {
355    let (conversation_id, participant_id, generation) = match value {
356        crate::wire::Retired::Enrollment {
357            request,
358            participant_id,
359            retired_generation,
360        } => (
361            request.conversation_id,
362            *participant_id,
363            *retired_generation,
364        ),
365        crate::wire::Retired::Participant {
366            request,
367            retired_generation,
368        } => {
369            let (conversation_id, participant_id) = participant_reference_identity(request);
370            (conversation_id, participant_id, *retired_generation)
371        }
372    };
373    aggregate.binding = ClientBindingState::Left {
374        conversation_id,
375        participant_id,
376        generation,
377    };
378    aggregate
379        .detach_replay
380        .apply_retired(conversation_id, participant_id, generation);
381}
382
383const fn participant_reference_identity(
384    request: &crate::wire::ParticipantReferenceEnvelope,
385) -> (u64, u64) {
386    match request {
387        crate::wire::ParticipantReferenceEnvelope::CredentialAttach(value) => {
388            (value.conversation_id, value.participant_id)
389        }
390        crate::wire::ParticipantReferenceEnvelope::Detach(value) => {
391            (value.conversation_id, value.participant_id)
392        }
393        crate::wire::ParticipantReferenceEnvelope::ParticipantAck(value) => {
394            (value.conversation_id, value.participant_id)
395        }
396        crate::wire::ParticipantReferenceEnvelope::Leave(value) => {
397            (value.conversation_id, value.participant_id)
398        }
399        crate::wire::ParticipantReferenceEnvelope::MarkerAck(value) => {
400            (value.conversation_id, value.participant_id)
401        }
402        crate::wire::ParticipantReferenceEnvelope::RecordAdmission(value) => {
403            (value.conversation_id, value.participant_id)
404        }
405    }
406}
liminal_protocol/client/inbound.rs

liminal_protocol/client/
inbound.rs
