frame_conv/handle/
state.rs

1//! The handle's state: one enrolled participant in one conversation, its
2//! inboxes, exchange registry, cursor tracker, and anomaly surface.
3
4use std::collections::{HashMap, VecDeque};
5use std::time::Duration;
6
7use frame_core::error::FailureReason;
8use liminal_protocol::wire::Generation;
9use liminal_sdk::remote::RemoteParticipantHandle;
10
11use crate::anomaly::{Anomaly, AnomalyCounters};
12use crate::envelope::Envelope;
13use crate::id::{ConversationId, ConversationSeq, CorrelationId, ParticipantRef};
14use crate::outcome::{DepartReason, JoinGrant};
15use crate::seam::StoreAdapter;
16use crate::store::ResumeStore;
17use crate::track::{CursorTracker, ExchangeRegistry};
18
19/// A reply record held for its requester, envelope decoded but payload not
20/// yet validated against the caller's typed message.
21#[derive(Debug)]
22pub(crate) struct HeldReply {
23    pub(crate) envelope: Envelope,
24    pub(crate) responder: ParticipantRef,
25    pub(crate) seq: ConversationSeq,
26}
27
28/// An inbound request record awaiting [`super::state::ConversationHandle::next_request`].
29#[derive(Debug)]
30pub(crate) struct HeldRequest {
31    pub(crate) envelope: Envelope,
32    pub(crate) requester: ParticipantRef,
33    pub(crate) seq: ConversationSeq,
34}
35
36/// A queued subscription item before typed payload decode. `Clone` because
37/// lifecycle and gap items fan into BOTH pattern inboxes (subscription and
38/// pub/sub) — each consumption surface presents a complete stream.
39#[derive(Debug, Clone)]
40pub(crate) enum QueuedItem {
41    Event {
42        envelope: Envelope,
43        publisher: ParticipantRef,
44        seq: ConversationSeq,
45    },
46    Joined {
47        peer: ParticipantRef,
48        seq: ConversationSeq,
49    },
50    Departed {
51        peer: ParticipantRef,
52        seq: ConversationSeq,
53        reason: DepartReason,
54    },
55    Failed {
56        peer: ParticipantRef,
57        seq: ConversationSeq,
58        failure: FailureReason,
59    },
60    Compacted {
61        seq: ConversationSeq,
62    },
63    Gap {
64        expected: ConversationSeq,
65        observed: ConversationSeq,
66    },
67}
68
69/// A peer death observed while exchanges may be open (F-3a R2's
70/// responder-failure wall; no live producer at liminal 0.3.0 — ASK-4).
71#[derive(Debug, Clone)]
72pub(crate) struct DeathNote {
73    pub(crate) peer: ParticipantRef,
74    pub(crate) failure: FailureReason,
75    pub(crate) seq: ConversationSeq,
76}
77
78/// One enrolled participant in one conversation: the transport-agnostic
79/// entry to the conversation patterns (design D4 — nothing in this type or
80/// its methods names a transport; the embedded transport arrives later
81/// behind this same handle).
82///
83/// A handle is single-owner and its pattern calls take `&mut self`; run one
84/// handle per participant, one participant per thread. The handle is `Send`
85/// so it may be moved into a thread after construction.
86pub struct ConversationHandle<S: ResumeStore> {
87    pub(crate) sdk: RemoteParticipantHandle<StoreAdapter<S>>,
88    pub(crate) conversation: ConversationId,
89    pub(crate) me: ParticipantRef,
90    pub(crate) grant: JoinGrant,
91    pub(crate) generation: Generation,
92    pub(crate) answer_window: Duration,
93    pub(crate) exchanges: ExchangeRegistry,
94    pub(crate) replies: HashMap<CorrelationId, HeldReply>,
95    pub(crate) requests_inbox: VecDeque<HeldRequest>,
96    pub(crate) events_inbox: VecDeque<QueuedItem>,
97    pub(crate) publications_inbox: VecDeque<QueuedItem>,
98    /// Last publication-record sequence presented through
99    /// `next_publication` — the O(1) exactly-once dedup state (F-3b R1's
100    /// strong branch; no set, no window).
101    pub(crate) last_presented_publication: Option<u64>,
102    pub(crate) deaths: Vec<DeathNote>,
103    pub(crate) anomalies: Vec<Anomaly>,
104    pub(crate) counters: AnomalyCounters,
105    pub(crate) tracker: CursorTracker,
106    pub(crate) attached: bool,
107}
108
109impl<S: ResumeStore> std::fmt::Debug for ConversationHandle<S> {
110    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111        formatter
112            .debug_struct("ConversationHandle")
113            .field("conversation", &self.conversation)
114            .field("participant", &self.me)
115            .field("attached", &self.attached)
116            .finish_non_exhaustive()
117    }
118}
119
120impl<S: ResumeStore> ConversationHandle<S> {
121    /// The conversation this handle is enrolled in.
122    #[must_use]
123    pub const fn conversation(&self) -> ConversationId {
124        self.conversation
125    }
126
127    /// This participant's substrate-minted identity.
128    #[must_use]
129    pub const fn participant(&self) -> ParticipantRef {
130        self.me
131    }
132
133    /// The current join grant: identity, attach credential, and generation.
134    /// The credential rotates on every resume; callers persisting the grant
135    /// must persist the CURRENT one.
136    #[must_use]
137    pub const fn grant(&self) -> &JoinGrant {
138        &self.grant
139    }
140
141    /// Whether the handle's connection is still serviceable. Once false,
142    /// every operation fails typed; the recovery path is
143    /// [`resume`](Self::resume) from the caller's persisted state.
144    #[must_use]
145    pub const fn attached(&self) -> bool {
146        self.attached
147    }
148
149    /// Drains the queued typed anomalies (R2's anomaly home, queue half).
150    /// Counters are unaffected by draining.
151    pub fn drain_anomalies(&mut self) -> Vec<Anomaly> {
152        std::mem::take(&mut self.anomalies)
153    }
154
155    /// The named, readable anomaly counters (R2's anomaly home, counter
156    /// half).
157    #[must_use]
158    pub const fn anomaly_counters(&self) -> AnomalyCounters {
159        self.counters
160    }
161}
frame_conv/handle/state.rs

frame_conv/handle/
state.rs
