frame_conv/handle/
attach.rs

1//! Opening, joining, and resuming conversations: enrollment (implicit
2//! create — F-0c assertion 1), and restore-then-reattach resume (F-0c §R4).
3
4use std::collections::{HashMap, VecDeque};
5
6use liminal_protocol::wire::{
7    AttachAttemptToken, AttachSecret, ClientRequest, CredentialAttachRequest, EnrollmentRequest,
8    EnrollmentToken, Generation, ServerValue,
9};
10use liminal_sdk::ConnectionPoolConfig;
11use liminal_sdk::remote::{RemoteConfig, RemoteParticipantError, RemoteParticipantHandle};
12
13use crate::config::BusAttachment;
14use crate::error::AttachError;
15use crate::id::{ConversationId, ParticipantRef};
16use crate::outcome::{JoinCredential, JoinGrant};
17use crate::seam::{StoreAdapter, classify_refusal, mint_token16};
18use crate::store::ResumeStore;
19use crate::track::{CursorTracker, ExchangeRegistry};
20
21use super::pump::SubmitFailure;
22use super::state::ConversationHandle;
23
24/// The participant surface still rides the channel-era remote configuration,
25/// which requires a channel name and a string conversation name the
26/// participant path never uses (recorded upstream ASK-3 — F-0c-FINDINGS
27/// §R2 recipe wart, re-measured at published 0.3.0). These fillers configure
28/// nothing and are deliberately NOT caller-facing configuration.
29const ASK3_UNUSED_CHANNEL: &str = "frame-conv-ask3-unused";
30const ASK3_UNUSED_CONVERSATION: &str = "frame-conv-ask3-unused";
31
32pub(super) fn connect(attachment: &BusAttachment) -> Result<RemoteConfig, AttachError> {
33    let config = RemoteConfig::new(
34        attachment.endpoint.clone(),
35        ASK3_UNUSED_CHANNEL,
36        ASK3_UNUSED_CONVERSATION,
37        ConnectionPoolConfig::new(
38            attachment.session_capacity,
39            attachment.operation_window_millis,
40            attachment.inbound_buffer,
41        ),
42    )
43    .map_err(|error| AttachError::Endpoint {
44        detail: format!("bus configuration rejected: {error}"),
45    })?;
46    config.connect_tcp().map_err(|error| AttachError::Endpoint {
47        detail: format!("bus endpoint unreachable: {error}"),
48    })
49}
50
51pub(super) fn map_restore_error(error: RemoteParticipantError) -> AttachError {
52    match error {
53        RemoteParticipantError::ResumeDecode(_)
54        | RemoteParticipantError::ResumeRestore(_)
55        | RemoteParticipantError::ResumeEncode(_) => AttachError::ResumeRejected {
56            detail: format!("{error}"),
57        },
58        RemoteParticipantError::Storage(_) => AttachError::Store {
59            detail: format!("{error}"),
60        },
61        other => AttachError::Protocol {
62            detail: format!("{other}"),
63        },
64    }
65}
66
67pub(crate) fn map_submit_failure(failure: SubmitFailure) -> AttachError {
68    match failure {
69        SubmitFailure::RecordRefused { detail, .. } | SubmitFailure::State { detail } => {
70            AttachError::Protocol { detail }
71        }
72        SubmitFailure::TransportLost { detail } | SubmitFailure::ConnectionFate { detail } => {
73            AttachError::ConnectionLost { detail }
74        }
75        SubmitFailure::AnswerTimeout { waited } => AttachError::AnswerTimeout { waited },
76        SubmitFailure::InboundRefused { detail } => AttachError::Protocol { detail },
77    }
78}
79
80impl<S: ResumeStore> ConversationHandle<S> {
81    /// Builds a handle whose identity fields are placeholders; every
82    /// constructor replaces them from a decoded binding receipt (or, on
83    /// the resumed-leave path, from the caller's persisted grant) before
84    /// the handle escapes.
85    pub(super) fn bare(
86        sdk: RemoteParticipantHandle<StoreAdapter<S>>,
87        conversation: ConversationId,
88        attachment: &BusAttachment,
89    ) -> Self {
90        Self {
91            sdk,
92            conversation,
93            me: ParticipantRef::from_wire(0),
94            grant: JoinGrant {
95                conversation,
96                participant: ParticipantRef::from_wire(0),
97                credential: JoinCredential::from_bytes([0; 32]),
98                generation: Generation::ONE.get(),
99            },
100            generation: Generation::ONE,
101            answer_window: attachment.answer_window,
102            exchanges: ExchangeRegistry::default(),
103            replies: HashMap::new(),
104            requests_inbox: VecDeque::new(),
105            events_inbox: VecDeque::new(),
106            publications_inbox: VecDeque::new(),
107            last_presented_publication: None,
108            deaths: Vec::new(),
109            anomalies: Vec::new(),
110            counters: crate::anomaly::AnomalyCounters::default(),
111            tracker: CursorTracker::default(),
112            attached: true,
113        }
114    }
115
116    /// Opens a NEW conversation: mints its identity and enrolls this
117    /// participant as its first member. Creation is implicit at first
118    /// enrollment (F-0c assertion 1 — the substrate offers no explicit
119    /// create and no create-versus-join testimony; the minting policy is
120    /// design §2.4's, carried until a server-issued binding authority
121    /// exists).
122    ///
123    /// # Errors
124    ///
125    /// Returns typed endpoint, refusal, transport, or answer-window
126    /// failures.
127    pub fn open(attachment: &BusAttachment, store: S) -> Result<(Self, JoinGrant), AttachError> {
128        Self::join(attachment, ConversationId::mint(), store)
129    }
130
131    /// Joins an existing conversation (or implicitly creates it — the
132    /// substrate cannot distinguish; see [`open`](Self::open)). Joining is
133    /// membership-forward: history before the join is neither delivered nor
134    /// ackable (the F-0c late-joiner finding).
135    ///
136    /// # Errors
137    ///
138    /// Returns typed endpoint, refusal, transport, or answer-window
139    /// failures.
140    pub fn join(
141        attachment: &BusAttachment,
142        conversation: ConversationId,
143        store: S,
144    ) -> Result<(Self, JoinGrant), AttachError> {
145        let config = connect(attachment)?;
146        let sdk = RemoteParticipantHandle::new(&config, StoreAdapter::new(store))
147            .map_err(map_restore_error)?;
148        let mut handle = Self::bare(sdk, conversation, attachment);
149        let answer = handle
150            .submit(ClientRequest::Enrollment(EnrollmentRequest {
151                conversation_id: conversation.to_wire(),
152                enrollment_token: EnrollmentToken::new(mint_token16()),
153            }))
154            .map_err(map_submit_failure)?;
155        match answer {
156            ServerValue::EnrollBound(receipt) => {
157                let me = ParticipantRef::from_wire(receipt.participant_id());
158                let generation = receipt.origin_binding_epoch().capability_generation;
159                let grant = JoinGrant {
160                    conversation,
161                    participant: me,
162                    credential: JoinCredential::from_bytes(receipt.attach_secret().into_bytes()),
163                    generation: generation.get(),
164                };
165                handle.me = me;
166                handle.generation = generation;
167                handle.grant = grant.clone();
168                Ok((handle, grant))
169            }
170            other => {
171                let (class, detail) = classify_refusal(&other);
172                Err(AttachError::Refused { class, detail })
173            }
174        }
175    }
176
177    /// Resumes a previously joined participant: restores the aggregate from
178    /// the caller's persisted canonical state, then re-attaches with the
179    /// grant's credential. Resume is restore-then-reattach, never restore
180    /// alone; the credential ROTATES and the returned grant carries the new
181    /// one — persist it (F-0c §R4, proven at published 0.3.0).
182    ///
183    /// On success the substrate replays the unacked obligation window from
184    /// the committed cursor — including records admitted while this
185    /// participant was dead — which arrives through the normal typed
186    /// surfaces ([`next_event`](Self::next_event) and friends).
187    ///
188    /// # Errors
189    ///
190    /// Returns typed endpoint, resume-state, refusal, transport, or
191    /// answer-window failures.
192    pub fn resume(
193        attachment: &BusAttachment,
194        grant: &JoinGrant,
195        canonical_state: &[u8],
196        store: S,
197    ) -> Result<(Self, JoinGrant), AttachError> {
198        let generation =
199            Generation::new(grant.generation).ok_or_else(|| AttachError::ResumeRejected {
200                detail: "grant carries the forbidden zero generation".to_owned(),
201            })?;
202        let config = connect(attachment)?;
203        let sdk =
204            RemoteParticipantHandle::restore(&config, StoreAdapter::new(store), canonical_state)
205                .map_err(map_restore_error)?;
206        let mut handle = Self::bare(sdk, grant.conversation, attachment);
207        handle.me = grant.participant;
208        handle.generation = generation;
209        handle.grant = grant.clone();
210        let answer = handle
211            .submit(ClientRequest::CredentialAttach(CredentialAttachRequest {
212                conversation_id: grant.conversation.to_wire(),
213                participant_id: grant.participant.to_wire(),
214                capability_generation: generation,
215                attach_secret: AttachSecret::new(grant.credential.to_bytes()),
216                attach_attempt_token: AttachAttemptToken::new(mint_token16()),
217                accept_marker_delivery_seq: None,
218            }))
219            .map_err(map_submit_failure)?;
220        match answer {
221            ServerValue::AttachBound(receipt) => {
222                let rotated_generation = receipt.origin_binding_epoch().capability_generation;
223                let rotated = JoinGrant {
224                    conversation: grant.conversation,
225                    participant: grant.participant,
226                    credential: JoinCredential::from_bytes(receipt.attach_secret().into_bytes()),
227                    generation: rotated_generation.get(),
228                };
229                handle.generation = rotated_generation;
230                handle.grant = rotated.clone();
231                Ok((handle, rotated))
232            }
233            other => {
234                let (class, detail) = classify_refusal(&other);
235                Err(AttachError::Refused { class, detail })
236            }
237        }
238    }
239}
frame_conv/handle/attach.rs

frame_conv/handle/
attach.rs
