lash/
recoverable_chat.rs

1//! Host-facing recoverable-chat observation seam.
2//!
3//! Lash owns the observation cursor, replay-gap, redelivery identity, and
4//! terminal-replacement contract. Hosts own authorization, product events,
5//! transcript presentation, and cancellation controls.
6
7use std::collections::BTreeSet;
8use std::pin::Pin;
9use std::task::{Context, Poll};
10
11use futures_util::Stream;
12use lash_core::{
13    LiveReplayGap, SessionCursor, SessionObservationEvent, SessionObservationEventPayload,
14    SessionReadView,
15};
16
17use crate::Result;
18use crate::session::{ObservableSession, SessionObservationStream, SessionObservationStreamItem};
19
20/// Subscription-scoped at-least-once delivery identity for one Lash
21/// observation event.
22///
23/// Hosts retain this identity only while the current live-replay incarnation is
24/// known to be continuous. Re-delivery of the same identity must not create
25/// another row. A [`RecoverableChatUpdate::ReplayGap`] ends that continuity:
26/// the replacement snapshot is authoritative and cursor identities retained
27/// from before the gap must be discarded because an in-memory replay store can
28/// reuse them after restart.
29#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
30pub struct RecoverableChatEventId {
31    pub session_id: String,
32    pub cursor: String,
33}
34
35impl RecoverableChatEventId {
36    fn from_event(event: &SessionObservationEvent) -> Self {
37        Self {
38            session_id: event.session_id.clone(),
39            cursor: event.cursor.to_string(),
40        }
41    }
42}
43
44/// One authoritative materialization paired with the cursor at which it was
45/// captured.
46#[derive(Clone, Debug)]
47pub struct RecoverableChatSnapshot {
48    pub read_view: SessionReadView,
49    pub cursor: SessionCursor,
50}
51
52impl RecoverableChatSnapshot {
53    pub fn capture(observable: &ObservableSession) -> Self {
54        let observation = observable.current_observation();
55        Self {
56            read_view: observation.read_view,
57            cursor: observation.cursor,
58        }
59    }
60}
61
62/// Update yielded by [`RecoverableChatSubscription`].
63#[derive(Clone, Debug)]
64pub enum RecoverableChatUpdate {
65    /// A provisional or lifecycle observation with stable redelivery identity.
66    Event {
67        id: RecoverableChatEventId,
68        event: std::sync::Arc<SessionObservationEvent>,
69    },
70    /// The requested cursor fell outside bounded replay. Replace the
71    /// projection from `snapshot`, persist `gap.latest_cursor`, and continue
72    /// consuming this same stream.
73    ReplayGap {
74        snapshot: RecoverableChatSnapshot,
75        gap: LiveReplayGap,
76    },
77    /// A terminal commit replaces provisional state with this authoritative
78    /// snapshot. The committed event is retained for remote encoding and
79    /// tracing, but the snapshot is the transcript authority.
80    TerminalReplacement {
81        id: RecoverableChatEventId,
82        event: std::sync::Arc<SessionObservationEvent>,
83        snapshot: RecoverableChatSnapshot,
84    },
85}
86
87/// Recoverable Lash observation stream for chat hosts.
88///
89/// Dropping this stream only disconnects observation. It never requests
90/// cancellation; cancellation remains an explicit call through
91/// [`TurnWorkDriver`](crate::TurnWorkDriver).
92pub struct RecoverableChatSubscription {
93    inner: SessionObservationStream,
94    applied: BTreeSet<RecoverableChatEventId>,
95}
96
97impl RecoverableChatSubscription {
98    pub(crate) fn new(inner: SessionObservationStream) -> Self {
99        Self {
100            inner,
101            applied: BTreeSet::new(),
102        }
103    }
104
105    /// Seed identities already applied by the host projection in this replay
106    /// store incarnation.
107    ///
108    /// This makes same-incarnation reconnect redelivery idempotent even when
109    /// the projection cursor intentionally trails individual applied events.
110    /// Do not persist these identities across a replay gap or process restart;
111    /// the subscription clears them when it emits
112    /// [`RecoverableChatUpdate::ReplayGap`].
113    pub fn with_applied_event_ids(
114        mut self,
115        ids: impl IntoIterator<Item = RecoverableChatEventId>,
116    ) -> Self {
117        self.applied.extend(ids);
118        self
119    }
120
121    pub fn cursor(&self) -> &SessionCursor {
122        self.inner.cursor()
123    }
124}
125
126impl Stream for RecoverableChatSubscription {
127    type Item = Result<RecoverableChatUpdate>;
128
129    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
130        loop {
131            let item = match Pin::new(&mut self.inner).poll_next(cx) {
132                Poll::Pending => return Poll::Pending,
133                Poll::Ready(None) => return Poll::Ready(None),
134                Poll::Ready(Some(Err(err))) => return Poll::Ready(Some(Err(err))),
135                Poll::Ready(Some(Ok(item))) => item,
136            };
137            match item {
138                SessionObservationStreamItem::Gap { observation, gap } => {
139                    self.applied.clear();
140                    return Poll::Ready(Some(Ok(RecoverableChatUpdate::ReplayGap {
141                        snapshot: RecoverableChatSnapshot {
142                            read_view: observation.read_view,
143                            cursor: observation.cursor,
144                        },
145                        gap,
146                    })));
147                }
148                SessionObservationStreamItem::Event(event) => {
149                    let id = RecoverableChatEventId::from_event(&event);
150                    if !self.applied.insert(id.clone()) {
151                        continue;
152                    }
153                    if let SessionObservationEventPayload::Committed { read_view } = &event.payload
154                    {
155                        self.applied.clear();
156                        self.applied.insert(id.clone());
157                        return Poll::Ready(Some(Ok(RecoverableChatUpdate::TerminalReplacement {
158                            id,
159                            snapshot: RecoverableChatSnapshot {
160                                read_view: read_view.clone(),
161                                cursor: event.cursor.clone(),
162                            },
163                            event,
164                        })));
165                    }
166                    return Poll::Ready(Some(Ok(RecoverableChatUpdate::Event { id, event })));
167                }
168            }
169        }
170    }
171}
172
173impl ObservableSession {
174    /// Capture one authoritative snapshot before attaching live observation.
175    pub fn recoverable_chat_snapshot(&self) -> RecoverableChatSnapshot {
176        RecoverableChatSnapshot::capture(self)
177    }
178
179    /// Resume a recoverable chat observation from an authoritative snapshot's
180    /// cursor.
181    pub fn subscribe_recoverable_chat(&self, cursor: SessionCursor) -> RecoverableChatSubscription {
182        RecoverableChatSubscription::new(self.subscribe_and_recover(cursor))
183    }
184}
lash/recoverable_chat.rs

lash/
recoverable_chat.rs
